+ @Bean
+ ApplicationHealthIndicator applicationHealthIndicator(
+ KafkaListenerEndpointRegistry registry,
+ KafkaProperties properties)
+ {
+ return new ApplicationHealthIndicator(
+ properties.getConsumer().getGroupId(),
+ registry);
+ }
+
+ @Bean
+ public ProducerFactory<String, Object> producerFactory(
+ KafkaProperties properties)
+ {
+ return new DefaultKafkaProducerFactory<>(
+ properties.getProducer().buildProperties(),
+ new StringSerializer(),
+ new DelegatingByTypeSerializer(
+ Map.of(
+ byte[].class, new ByteArraySerializer(),
+ MessageAddNumber.class, new JsonSerializer<>(),
+ MessageCalculateSum.class, new JsonSerializer<>())));
+ }
+
+ @Bean
+ public KafkaTemplate<String, Object> kafkaTemplate(
+ ProducerFactory<String, Object> producerFactory)
+ {
+ return new KafkaTemplate<>(producerFactory);
+ }