add producer consumer interceptors

[vvuibert]

Add producer and consumer interceptors that can intercept messages at different points on producer and consumer in order to trace the path of individual messages across the cluster.

• inspired in KIP-42
• intercepts produced messages before they are sent to the brokers
• intercepts consumed messages before are returned to the client
• does not intercept the broker's acknowledgement
• does not intercept when the consumer commits

[d1egoaz]

ping @dasch to see if he can schedule some time to review this, thanks!

[dasch]

We already use ActiveSupport::Notifications to provide granular pub/sub hooks – what is the use case for introducing this new mechanism?

[vvuibert]

@dasch with interceptors we can add the span and trace ID to the message headers so that traces emitted by producers can be linked by the consumer.
Interceptor API will allow mutate the records to support the ability to add metadata to a message for auditing/end-to-end monitoring.

[dasch]

@vvuibert ah OK, that’s a good use case, and something we need at Zendesk as well. I’ll review this then.

[dasch]

In general, I think this implementation follows the Java design too closely. ruby-kafka tries to follow Ruby idioms to the degree possible.

To me, that means dropping a class hierarchy in favor of either a simple contract with a couple of expected methods on the objects being passed or, better yet, a simplified API consisting of only callables.

Specifically, unless there’s an actual need for a close method, we could do away with having separate concepts for producer and consumer interceptors, instead just expecting the passed objects to have a call method. This allows passing in arbitrary lambdas, procs, method objects, etc., and is the typical way to do it in Ruby.

We could take that a step further and allow setting these in a simple DSL, e.g.

producer = kafka.producer(...)

producer.on_send do |message|
  # mutate message here!
end

consumer = kafka.consumer(...)

consumer.on_batch do |batch|
  # do you stuff here!
end

consumer.on_message do |message|
  # simplified API for per-message stuff!
end

[dasch]

I still think we need to simplify the API and have just a single container for the interceptor lists.

[dasch]

Awesome! Just a little spit and polish and I’ll be happy to merge!

Thanks for the contribution!

[dasch]

Thanks!

[d1egoaz]

Thanks @dasch for the detailed review!

I wonder what are the plans for an upcoming release that includes this change.
Is ruby-kafka using some kind of release cycle?

I see there are some issue with the concurrent dependency #835 that might need to be addressed first before a new release.

We'll keep and 👀 on new releases!

Thanks

[dasch]

Yeah, hoping to get that resolved before issuing a new release. There’s no firm cycle.

[dasch]

Addressing that issue would certainly expedite a new release :D