How to mutate ProducerRecord in ProducerInterceptor when used via KafkaTemplate?

[pw-lehre]

Suppose one has a KafkaTemplate configuration like this:

var properties = HashMap<String, Object>();
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
var producerFactory = new DefaultKafkaProducerFactory<>(properties);
var kafkaTemplate = new KafkaTemplate<>(producerFactory);

kafkaTemplate .setProducerInterceptor(myCustomInterceptor);

According to the ProducerInterceptor.onSend documentation I can return a new mutated producer record, but I cannot get this to work.

ProducerInterceptor.onSend(...)

This method is allowed to modify the record, in which case, the new record will be returned.

To me it seems that the interceptor return value is not used at all.

How is this supposed to be used?

[garyrussell]

This is a bug; cc/ @sobychacko

In the meantime, you can add the interceptor class name to the producer factory configuration properties instead of to the template.

[Wzy19930507]

Hi, @garyrussell
After fixes this bug,
is it feasible to use org.apache.kafka.clients.producer.internals.ProducerInterceptors instead of ProducerInterceptor?

[garyrussell]

It seems odd that they created a new class for that; I would have preferred

public class CompositeProducerInterceptor<K, V> implements ProducerInterceptor<K, V> { ... }

That's the pattern we usually use here. I would rather not clutter the KafkaTemplate with this additional ProducerInterceptors property.

ProducerInterceptor was added to the template to allow injection of Spring-managed interceptors (which is not possible via config properties).

[Wzy19930507]

much better, thank you for your answer :)