Document how to wire other beans in Spring Kafka producer/consumer interceptor

[nitenA]

Hi ,

Please let me know how to auto wire beans with spring Kafka producer/consumer interceptor. Interceptor works fine and getting invoked on each commit/ack.

I would like to take it further by calling custom bean to record that into database.

Also how to get application configurations within this interceptor.

I tried all possible ways of making it as component and registering to spring interceptor registry but no luck.

[garyrussell]

Did you not read the instructions?

🙅 "Please DO NOT Raise an Issue" Cases

• Question
  STOP!! Please ask questions about how to use something, or to understand why something isn't
  working as you expect it to, on Stack Overflow using the spring-kafka tag.

You cannot; they are not managed by Spring; they are managed by Kafka itself, which knows nothing about Spring.

[garyrussell]

You could write a customer producer factory that decorates the producer(s) created by the DefaultKafkaProducerFactory, but that's not the same as a native interceptor.

[mawek]

I think it should be possible - interceptors implement Configurable interface which means that if you put some spring bean to your properties map when creating KafkaTemplate/ConsumerFactory, you can access it within configure method of interceptor (and assign it to local variable or whatever)

[artembilan]

Oh! That's good catch, @mawek !

Deserves to be documented 😄

Feel free to contribute your observation!

Thank you!

[nitenA]

Let me try and let you know if this works. Thanns @mawek for suggestion and it will be great help if works out.

[garyrussell]

Not to be pedantic, but this is not exactly "auto wiring"; it is a mechanism to provide user code access to Spring-managed beans to allow it to do the wiring manually 😄

Also, this will only work if you create the configs yourself (or use Boot's KafkaProperties.getConsumer().buildProperties() and then add the bean(s) afterwards, before creating the factory.

[lucek091]

@mawek @artembilan @nitenA
Do you have some code snippet or github how did you resolve this problem?

[garyrussell]

@lucek091 Here you go...

@SpringBootApplication
public class Kgh897Application {

	public static void main(String[] args) {
		SpringApplication.run(Kgh897Application.class, args);
	}

	@Bean
	public ConsumerFactory<?, ?> kafkaConsumerFactory(KafkaProperties properties, Foo foo) {
		Map<String, Object> consumerProperties = properties.buildConsumerProperties();
		consumerProperties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName());
		consumerProperties.put("foo.bean", foo);
		return new DefaultKafkaConsumerFactory<>(consumerProperties);
	}

	@Bean
	public ProducerFactory<?, ?> kafkaProducerFactory(KafkaProperties properties, Foo foo) {
		Map<String, Object> producerProperties = properties.buildProducerProperties();
		producerProperties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
		producerProperties.put("foo.bean", foo);
		DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(producerProperties);
		String transactionIdPrefix = properties.getProducer()
				.getTransactionIdPrefix();
		if (transactionIdPrefix != null) {
			factory.setTransactionIdPrefix(transactionIdPrefix);
		}
		return factory;
	}

	@Bean
	public Foo foo() {
		return new Foo();
	}

	@KafkaListener(id = "kgk897", topics = "kgh897")
	public void listen(String in) {
		System.out.println("Received " + in);
	}

	@Bean
	public ApplicationRunner runner(KafkaTemplate<String, String> template) {
		return args -> template.send("kgh897", "test");
	}

	@Bean
	public NewTopic topic() {
		return new NewTopic("kgh897", 1, (short) 1);
	}

}

public class Foo {

	public void foo(String what) {
		System.out.println(what + " in my foo bean");
	}

}

public class MyConsumerInterceptor implements ConsumerInterceptor<String, String> {

	private Foo foo;

	@Override
	public void configure(Map<String, ?> configs) {
		this.foo = (Foo) configs.get("foo.bean");
	}

	@Override
	public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
		this.foo.foo("consumer interceptor");
		return records;
	}

	@Override
	public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
	}

	@Override
	public void close() {
	}

}

public class MyProducerInterceptor implements ProducerInterceptor<String, String> {

	private Foo foo;

	@Override
	public void configure(Map<String, ?> configs) {
		this.foo = (Foo) configs.get("foo.bean");
	}

	@Override
	public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
		this.foo.foo("producer interceptor");
		return record;
	}

	@Override
	public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
	}

	@Override
	public void close() {
	}

}

spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest

and

producer interceptor in my foo bean
consumer interceptor in my foo bean
Received test