Make asynchronous emitters have the same (configurable?) policy when event queue is full

[leventov]

This issue narrows the idea of #7037.

There are many emitters (at least: AmbariMetricsEmitter, GraphiteEmitter, StatsDEmitter and KafkaEmitter and HttpPostEmitter, I haven't checked other) that use the same producer-consumer pattern for asynchronous emit: emit() pushes the event to some queue (one of the queues), and there is an asynchronous executor that retrieves events from the queue and sends them over network to emit.

AmbariMetricsEmitter and GraphiteEmitter use the same policy when the queue is full (they log a warning). But StatsDEmitter apparently silently discards new events when the queue is full (see NonBlockingStatsDClient code). KafkaEmitter discards new events, but increments "lost events" counts. HttpPostEmitter packs events in batches and drops the oldest batch when overwhelmed, simultaneously logging that (see HttpPostEmitter.limitBuffersToEmitSize() and limitFailedBuffersSize()).

I think all emitters should be similar in this regard. Probably event throttling policy should be configurable.

Related to #2868.

[justinborromeo]

Do you have any strong opinions on which policy would be the best to use across all the emitters?

[leventov]

In my opinion, dropping old events is better than dropping new ones (as StatsDEmitter and KafkaEmitter any maybe some other emitters currently do). Logging a warning or error alongside dropping events also makes sense.

[justinborromeo]

Also, what do you mean by "backpressure policy"? Does that just involve dropping the oldest events in queue?

[leventov]

Yes, that was not very precise to call it "backpressure" policy, because we probably never want actual backpressure in metrics emitter, because metrics are relatively unimportant to slow down queries. Rephrased to "event throttling" policy. Simple policies might be "drop oldest" and "drop newest". Some more sophisticated policies may change the event "level" (like logging level) when the queue is full, like stop emitting some relatively less important kinds of events. (Probably trying to return to the default level after some backoff time or when the queue becomes sufficiently empty).

[justinborromeo]

Looking at the code, I noticed that NonBlockingStatsDEmitter is a library class. Does it make sense to ignore it when refactoring since we can't change it or do you think that a custom implementation should be written?

[justinborromeo]

Opened a proposal (#7075) since the change adds a config.

[leventov]

I've created an issue in the repo of that library: DataDog/java-dogstatsd-client#71

[justinborromeo]

Does that correspond to what we're using? NonBlockingStatsDClient isn't a Datadog library.

[leventov]

@justinborromeo /~https://github.com/DataDog/java-dogstatsd-client/blob/de8c7982d208b1458e17eef81e4d18085d7fd9f5/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java don't we use this class?

[justinborromeo]

Sorry, I misread something. You are correct.

[github-actions]

This issue has been marked as stale due to 280 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If this issue is still
relevant, please simply write any comment. Even if closed, you can still revive the
issue at any time or discuss it on the dev@druid.apache.org list.
Thank you for your contributions.