GH-2049: Spring Managed Producer Interceptors #2112
Conversation
| /** | ||
| * Returns the producer interceptor associated with this template. | ||
| * @return {@link ProducerInterceptor} | ||
| * @since 3.0.x |
There was a problem hiding this comment.
Just 3.0
And Why do we need a getter?
If we do, perhaps you mean protected for extenders...
| @@ -3734,6 +3734,67 @@ Received test | |||
| ---- | |||
| ==== | |||
|
|
|||
| ==== Producer Interceptor Managed in Spring | |||
|
|
|||
| Staring with version 3.0.0, when it comes to a producer interceptor, you can let Spring manage it directly as a bean instead of providing the class name of the interceptor to the Apache Kafka producer configuration. | |||
There was a problem hiding this comment.
Small typo: "Starting"
| @@ -631,6 +644,9 @@ protected void closeProducer(Producer<K, V> producer, boolean inTx) { | |||
| if (this.micrometerHolder != null) { | |||
| sample = this.micrometerHolder.start(); | |||
| } | |||
| if (this.producerInterceptor != null) { | |||
| this.producerInterceptor.onSend(producerRecord); | |||
There was a problem hiding this comment.
I guess we don't need to catch an exception here; a user might want to use this to cancel a send for some reason.
There was a problem hiding this comment.
It looks like Kafka's ProducerInterceptors simply log the exception from individual interceptors and continue. See here. The one set on the KafkaTemplate can be used to cancel if need be. I was thinking maybe with a CompositeProducerInterceptor we can follow the same approach used in Kafka's ProducerInterceptors.
| @@ -657,6 +673,9 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final | |||
| final SettableListenableFuture<SendResult<K, V>> future, @Nullable Object sample) { | |||
|
|
|||
| return (metadata, exception) -> { | |||
| if (this.producerInterceptor != null) { | |||
| this.producerInterceptor.onAcknowledgement(metadata, exception); | |||
There was a problem hiding this comment.
Need to catch and log any exceptions here, then proceed; otherwise the future won't be completed.
|
It is not important if this misses Tuesday's M2 release. |
| record.topic(), record.partition())); | ||
| } | ||
| else { | ||
| CompositeProducerInterceptor.this.logger.warn(e, () -> "Error executing interceptor onSend callback"); |
There was a problem hiding this comment.
Maybe add interceptor.toString() to these log messages?
Allow producer interceptors to be managed by Spring so that they can be used in KafkaTemplate instead of providing the classname of the interceptor to Kafka configuraiton. Adding tests and docs. Resolves #2049
Adding a CompositeProducerInterceptor to allow multiple producer interceptors on KafkaTemplate.
Allow producer interceptors to be managed by Spring so that
they can be used in KafkaTemplate instead of providing the classname
of the interceptor to Kafka configuraiton.
Adding tests and docs.
Resolves #2049