Spring Kafka Template – Connect to Kafka Topic on Spring Boot Startup

I have implemented a basic Spring Boot Application which uses Spring Kafka. I want my producer to connect to the Kafka Topic before the first .send() is called but I can’t find a way to do so. Is that possible?

Logs to show that KafkaTemplate only connects to the Kafka Topic after I trigger the .send method at 16:12:44:

2021-11-24 16:12:12.602  INFO 63930 --- [           main] c.e.k.KafkaProducerExampleApplication    : The following profiles are active: dev
2021-11-24 16:12:13.551  INFO 63930 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
2021-11-24 16:12:13.559  INFO 63930 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
2021-11-24 16:12:13.559  INFO 63930 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.53]
2021-11-24 16:12:13.613  INFO 63930 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
2021-11-24 16:12:13.613  INFO 63930 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 974 ms
2021-11-24 16:12:13.989  INFO 63930 --- [           main] pertySourcedRequestMappingHandlerMapping : Mapped URL path [/v2/api-docs] onto method [springfox.documentation.swagger2.web.Swagger2Controller#getDocumentation(String, HttpServletRequest)]
2021-11-24 16:12:14.190  INFO 63930 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
2021-11-24 16:12:14.190  INFO 63930 --- [           main] d.s.w.p.DocumentationPluginsBootstrapper : Context refreshed
2021-11-24 16:12:14.207  INFO 63930 --- [           main] d.s.w.p.DocumentationPluginsBootstrapper : Found 1 custom documentation plugin(s)
2021-11-24 16:12:14.239  INFO 63930 --- [           main] s.d.s.w.s.ApiListingReferenceScanner     : Scanning for api listing references
2021-11-24 16:12:14.336  INFO 63930 --- [           main] c.e.k.KafkaProducerExampleApplication    : Started KafkaProducerExampleApplication in 7.055 seconds (JVM running for 7.341)
2021-11-24 16:12:44.550  INFO 63930 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2021-11-24 16:12:44.550  INFO 63930 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2021-11-24 16:12:44.551  INFO 63930 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 1 ms
2021-11-24 16:12:44.649  INFO 63930 --- [nio-8080-exec-1] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values:

Answer

With non-transactional producer (transactionIdPrefix is not supplied), when you first call KafkaTemplate.send, it will delegate to ProducerFactory to get a single instance of Producer. At this time, because there’s no a single instance of Producer before, ProducerFactory will create this one for you (that’s why you saw the log ProducerConfig : ProducerConfig values ...). This producer instance is now used/shared by all clients.

So if you want to create the above producer instance beforehand, you could directly call it on the ProducerFactory, e.g:

@Bean
public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory) {
        KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate(kafkaProducerFactory);
        kafkaProducerFactory.createProducer();
        return kafkaTemplate;
...

Advertisement

Answer