I am using Brave library https://github.com/openzipkin/brave for tracing and now I would like to use it also for Kafka consumer. I would like to avoid adding Spring Sleuth and leverage just Brave Kafka instrumentation https://github.com/openzipkin/brave/tree/master/instrumentation/kafka-clients.
For the Kafka consumer I use @KafkaListener. The code looks like this:
TestKafkaEndpoint.java
@Service
public class TestKafkaEndpoint {
@KafkaListener(topics = "myTestTopic", containerFactory = "testKafkaListenerContainerFactory")
public void procesMyRequest(@Payload final MyRequest request) {
// do some magic...
}
}
And configuration class TestKafkaConfig.java
@Configuration
@EnableKafka
@ComponentScan
public class TestKafkaConfig {
@Bean
public ConsumerFactory<String, MyRequest> testConsumerFactory() {
final Map<String, Object> consumerProperties = new HashMap<>();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka01-localhost:9092");
consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "TestGROUP");
return new DefaultKafkaConsumerFactory<>(consumerProperties, new StringDeserializer(), new JsonDeserializer<>(MyRequest.class));
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, MyRequest>> testKafkaListenerContainerFactory() {
final ConcurrentKafkaListenerContainerFactory<String, MyRequest> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(testConsumerFactory());
factory.getContainerProperties().setErrorHandler(new LoggingErrorHandler());
return factory;
}
But I do not know how to use KafkaConsumer when using Kafka factory or leverage the KafkaTracing. Does anyone has any experience with that and got it working?
I'm not familiar with it, but it looks like the TracingConsumer is a simple consumer wrapper: https://github.com/openzipkin/brave/blob/363ceb4c922305ffb4a68ac47dc152e1d15da0fb/instrumentation/kafka-clients/src/main/java/brave/kafka/clients/TracingConsumer.java#L69-L79
You should be able to create a subclass of DefaultKafkaConsumerFactory; override the createConsumer method(s) - the listener container uses...
this.consumer =
KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
this.consumerGroupId,
this.containerProperties.getClientId(),
KafkaMessageListenerContainer.this.clientIdSuffix,
consumerProperties);
... call super.createConsumer(...) and wrap it in a TracingConsumer.
If you are using 2.5.3 or later, you can add a ConsumerPostProcessor to the DKCF.
That's how sleuth does it: