FYI for future readers, the thing that made it work was indeed the code supplied by Tommaso. Specifically, creating a custom @Bean(“factory”), then using it in the @KafkaListener(containerFactory = “factory”).

Thanks a lot, because I had been on this issue for a loooong day.

Hi. But what exactly did you change? Just added that line: factory.getContainerProperties().setMicrometerEnabled(true); ?

@Bean(“factory”)
public ConcurrentKafkaListenerContainerFactory listenerFactory(ConsumerFactory consumerFactory) {
ConcurrentKafkaListenerContainerFactory factory =
new ConcurrentKafkaListenerContainerFactory();
factory.getContainerProperties().setObservationEnabled(true);
factory.getContainerProperties().setMicrometerEnabled(true);
factory.setConsumerFactory(consumerFactory);
returnfactory;
}

@Service
public class ListenerService {
private static final Logger LOG = LoggerFactory.getLogger(ListenerService.class);
@KafkaListener(id = “info”, topics = “${app.in.topic}”, containerFactory = “factory”)
public void onMessage(@Payload Info info,
@Header(name = KafkaHeaders.RECEIVED_KEY, required = false) Long key,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition) {
LOG.info(“Received(key={}, partition={}): {}”, key, partition, info);
}

}
Greetings

The library add trace id and span id headers to the messages

No. But I know that currently Spring team works on the simplifying it onside the Spring Kafka. With Spring Boot 3.2 you can use YAML/properties instead of the `setObservationEnabled` method.

Ok, thanks for that suggestion.

In order to achive it, we should override the default Spring Boot log format (e.g. the `logging.pattern.console` property). Maybe in the future, the Spring Boot Kafka autoconfiguration will do that.