[PROPOSAL] Add a Configurable Event Throttling Policy to Asynchronous Emitters

[justinborromeo]

Motivation

See the issue raised by @leventov (#7057).

Proposed changes

I think that out of the five Emitters listed in #7057, I would change all but StatsDEmitter. The code that handles enqueuing for the StatsDEmitter is in library class NonBlockingStatsDClient so there wouldn’t really be any way of changing it without rewriting the class.

The two approaches to refactoring:

1. In an ideal world, I'd create an abstract class QueueBasedEmitter which holds the event queue and implements int offerAndHandleFailure(event), and have the four emitters that write to queue (see list in Motivation) extend it. This method would try to offer the event to the queue. If the offering fails, events would be discarded from the queue (based on the configurable policy) until the offer succeeds (this approach assumes that the only reason for a offer() call failing is a full queue). The number of discarded events would be returned from the function so that the Emitter implementation can use it accordingly; for example, KafkaEmitter has a count of lost events that is periodically logged. However, I'm not sure if this is feasible since AmbariMetricsEmitter already extends AbstractTimelineMetricsSink and Java doesn't support multiple inheritance. It would be possible if AmbariMetricsEmitter was ignored in the refactor but that doesn't seem to be a very object-oriented approach.

2. I think the more feasible approach would be to create a helper class that accepts the event queue as a constructor argument and implements int offerAndHandleFailure(event). The helper class would be instantiated when the Emitter constructor is called. The helper would work in AmbariMetricsEmitter, KafkaEmitter, and GraphEmitter with minimal change since they all already use BlockingQueues. HttpPostEmitter can use the queue approach if its buffersToEmit Deque is converted to a BlockingQueue. I think this substitution would work since only addLast() and pollFirst() are called on the Deque.

With regard to configuration, I'm thinking of adding the following config parameters:

1. ...emitter.eventThrottlingPolicy [String]: specifies what policy to follow when the queue is full and a new event is offered (i.e. BlockingQueue#offer() times out)

• "dropOldest" (default)": discards oldest event from the queue
• "dropNewest": discards the event that was just offered

2. ...emitter.eventThrottlingLoggingFrequency [double from 0 to 1]: specifies how often an event being dropped should be logged. This policy is pretty naive as it assumes the offer failure frequency is constant. A better (but more challenging) approach might be to tune the logging rate based on the number of failures that have occurred in the past n seconds/minutes.

• 1: Log every dropped event
• 0.001: Log 1 in every 1000 dropped events
• 0: Don't log any dropped events

If anyone thinks that any of the above configs should be removed or modified, or another should be added, leave a comment below.

Rationale

I believe the helper-based approach to refactoring event emission is the best approach because it is the most object-oriented and avoids doing anything hacky with existing code. The one other option is to have each Emitter implementation implement its own handling of the event throttling policy but that doesn't really help with code reuse.

Operational impact

Likely not? It would be awesome if someone with more knowledge of users' use cases with metrics could comment on this.

Open questions

1. What's the best thing to log when an event is dropped? Count of dropped events, a serialized version of the event, or just the fact that an event has been discarded?
2. Does this change have an operational impact?

[leventov]

AmbariMetricsEmitter doesn't need to extend AbstractTimelineMetricsSink. This is actually a bad use of inheritance instead of composition. Its internal ConsumerRunnable can extend AbstractTimelineMetricsSink, or a separate private class.

I'm not sure we should introduce configurations before somebody actually needs them. One reason is that it's hard to change HttpPostEmitter to support policies different from "drop oldest batch", that is also not exactly equivalent to queue-based emitters's "drop oldest events". I think we may start with just dropping oldest events in queue-based emitters and reporting, for example, one 1000th drop (but making sure that we report the very first one).

[justinborromeo]

I'm looking at the Kafka metrics emitter and I'm kinda stumped about a design decision. The way it's currently implemented, it follows the pattern of all the other emitters where events are pushed to some queue then some thread dequeues events and does something with them. In this case, the thread sends the event to Kafka using the KafkaProducer. From what I understand, the Kafka producer has its own memory-bounded buffer where records are batched before actually being sent. From the javadoc:

The producer consists of a pool of buffer space that holds records that haven't yet been transmitted to the server as well as a background I/O thread that is responsible for turning these records into requests and transmitting them to the cluster.

If this is the case, is there any benefit of having the first queue if it's being buffered on the same machine at a later stage? From the original commit,

In terms of code maintenance, Using block.on.buffer.full is a little bit questionable solution. also, @b-slim mentioned, emit should never be blocking but using producer.send in emit has potential blocking issue because of max.block.ms. This is why I use another queue and using queue.add in emit.

However, if max.block.ms is set to 0, wouldn't it just fast-fail if the buffer is full?

[leventov]

@justinborromeo interesting. Does KafkaProducer support custom event dropping policies? It seems to me that we should treat it as we treat NonBlockingStatsDClient.

[justinborromeo]

I don't think it does. I think it just fails the message that's being added to the full queue.

[leventov]

Could you please open an issue in the Kafka library?

[justinborromeo]

Done: https://issues.apache.org/jira/browse/KAFKA-7953

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[stale]

This issue is no longer marked as stale.