[JAVA-29500] Moved article "Implementing Retry in Kafka Consumer" to spring-kafka (#15904)

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/Farewell.java

@@ -1,37 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-public class Farewell {
-
-    private String message;
-    private Integer remainingMinutes;
-
-    public Farewell() {
-    }
-
-    public Farewell(String message, Integer remainingMinutes) {
-        this.message = message;
-        this.remainingMinutes = remainingMinutes;
-    }
-
-    public String getMessage() {
-        return message;
-    }
-
-    public void setMessage(String message) {
-        this.message = message;
-    }
-
-    public Integer getRemainingMinutes() {
-        return remainingMinutes;
-    }
-
-    public void setRemainingMinutes(Integer remainingMinutes) {
-        this.remainingMinutes = remainingMinutes;
-    }
-
-    @Override
-    public String toString() {
-        return message + ". In " + remainingMinutes + "!";
-    }
-
-}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/Greeting.java

@@ -1,37 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-public class Greeting {
-
-    private String msg;
-    private String name;
-
-    public Greeting() {
-
-    }
-
-    public Greeting(String msg, String name) {
-        this.msg = msg;
-        this.name = name;
-    }
-
-    public String getMsg() {
-        return msg;
-    }
-
-    public void setMsg(String msg) {
-        this.msg = msg;
-    }
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    @Override
-    public String toString() {
-        return msg + ", " + name + "!";
-    }
-}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/KafkaConsumerConfig.java

@@ -1,151 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-import org.springframework.kafka.annotation.EnableKafka;
-import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
-import org.springframework.kafka.core.ConsumerFactory;
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
-import org.springframework.kafka.listener.ContainerProperties;
-import org.springframework.kafka.listener.DefaultErrorHandler;
-import org.springframework.kafka.support.converter.RecordMessageConverter;
-import org.springframework.kafka.support.converter.StringJsonMessageConverter;
-import org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper;
-import org.springframework.kafka.support.mapping.Jackson2JavaTypeMapper;
-import org.springframework.kafka.support.serializer.JsonDeserializer;
-import org.springframework.util.backoff.BackOff;
-import org.springframework.util.backoff.FixedBackOff;
-
-@EnableKafka
-@Configuration
-public class KafkaConsumerConfig {
-
-    @Value(value = "${spring.kafka.bootstrap-servers}")
-    private String bootstrapAddress;
-
-    @Value(value = "${kafka.backoff.interval}")
-    private Long interval;
-
-    @Value(value = "${kafka.backoff.max_failure}")
-    private Long maxAttempts;
-
-    public ConsumerFactory<String, String> consumerFactory(String groupId) {
-        Map<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
-        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "20971520");
-        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "20971520");
-        return new DefaultKafkaConsumerFactory<>(props);
-    }
-
-    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(String groupId) {
-        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
-        factory.setConsumerFactory(consumerFactory(groupId));
-        return factory;
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, String> fooKafkaListenerContainerFactory() {
-        return kafkaListenerContainerFactory("foo");
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, String> barKafkaListenerContainerFactory() {
-        return kafkaListenerContainerFactory("bar");
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, String> headersKafkaListenerContainerFactory() {
-        return kafkaListenerContainerFactory("headers");
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, String> partitionsKafkaListenerContainerFactory() {
-        return kafkaListenerContainerFactory("partitions");
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, String> longMessageKafkaListenerContainerFactory() {
-        return kafkaListenerContainerFactory("longMessage");
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, String> filterKafkaListenerContainerFactory() {
-        ConcurrentKafkaListenerContainerFactory<String, String> factory = kafkaListenerContainerFactory("filter");
-        factory.setRecordFilterStrategy(record -> record.value()
-          .contains("World"));
-        return factory;
-    }
-
-    public ConsumerFactory<String, Greeting> greetingConsumerFactory() {
-        Map<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "greeting");
-        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(Greeting.class));
-    }
-
-    @Bean
-    public ConcurrentKafkaListenerContainerFactory<String, Greeting> kafkaListenerContainerFactory() {
-        ConcurrentKafkaListenerContainerFactory<String, Greeting> factory = new ConcurrentKafkaListenerContainerFactory<>();
-        factory.setConsumerFactory(greetingConsumerFactory());
-        return factory;
-    }
-
-    @Bean
-    public RecordMessageConverter multiTypeConverter() {
-        StringJsonMessageConverter converter = new StringJsonMessageConverter();
-        DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
-        typeMapper.setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID);
-        typeMapper.addTrustedPackages("com.baeldung.spring.kafka");
-        Map<String, Class<?>> mappings = new HashMap<>();
-        mappings.put("greeting", Greeting.class);
-        mappings.put("farewell", Farewell.class);
-        typeMapper.setIdClassMapping(mappings);
-        converter.setTypeMapper(typeMapper);
-        return converter;
-    }
-
-    @Bean
-    public ConsumerFactory<String, Object> multiTypeConsumerFactory() {
-        HashMap<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id_test");
-        return new DefaultKafkaConsumerFactory<>(props);
-    }
-
-    @Bean
-    @Primary
-    public ConcurrentKafkaListenerContainerFactory<String, Object> greetingKafkaListenerContainerFactory() {
-        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
-        factory.setConsumerFactory(multiTypeConsumerFactory());
-        factory.setMessageConverter(multiTypeConverter());
-        factory.setCommonErrorHandler(errorHandler());
-        factory.getContainerProperties()
-          .setAckMode(ContainerProperties.AckMode.RECORD);
-        return factory;
-    }
-
-    @Bean
-    public DefaultErrorHandler errorHandler() {
-        BackOff fixedBackOff = new FixedBackOff(interval, maxAttempts);
-        DefaultErrorHandler errorHandler = new DefaultErrorHandler((consumerRecord, e) -> {
-            System.out.println(String.format("consumed record %s because this exception was thrown",consumerRecord.toString(),e.getClass().getName()));
-        }, fixedBackOff);
-        //Commented because of the test
-        //errorHandler.addRetryableExceptions(SocketTimeoutException.class,RuntimeException.class);
-        errorHandler.addNotRetryableExceptions(NullPointerException.class);
-        return errorHandler;
-    }
-
-}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/KafkaProducerConfig.java

@@ -1,67 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.support.serializer.JsonSerializer;
-
-@Configuration
-public class KafkaProducerConfig {
-
-    @Value(value = "${spring.kafka.bootstrap-servers}")
-    private String bootstrapAddress;
-
-    @Bean
-    public ProducerFactory<String, String> producerFactory() {
-        Map<String, Object> configProps = new HashMap<>();
-        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        configProps.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "20971520");
-
-        return new DefaultKafkaProducerFactory<>(configProps);
-    }
-
-    @Bean
-    public KafkaTemplate<String, String> kafkaTemplate() {
-        return new KafkaTemplate<>(producerFactory());
-    }
-
-    @Bean
-    public ProducerFactory<String, Greeting> greetingProducerFactory() {
-        Map<String, Object> configProps = new HashMap<>();
-        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
-        return new DefaultKafkaProducerFactory<>(configProps);
-    }
-
-    @Bean
-    public KafkaTemplate<String, Greeting> greetingKafkaTemplate() {
-        return new KafkaTemplate<>(greetingProducerFactory());
-    }
-
-    @Bean
-    public ProducerFactory<String, Object> multiTypeProducerFactory() {
-        Map<String, Object> configProps = new HashMap<>();
-        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
-        configProps.put(JsonSerializer.TYPE_MAPPINGS, "greeting:com.baeldung.spring.kafka.retrayable.Greeting, farewell:com.baeldung.spring.kafka.retrayable.Farewell");
-        return new DefaultKafkaProducerFactory<>(configProps);
-    }
-
-    @Bean
-    public KafkaTemplate<String, Object> multiTypeKafkaTemplate() {
-        return new KafkaTemplate<>(multiTypeProducerFactory());
-    }
-
-}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/KafkaTopicConfig.java

@@ -1,77 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.kafka.core.KafkaAdmin;
-
-@Configuration
-public class KafkaTopicConfig {
-
-    @Value(value = "${spring.kafka.bootstrap-servers}")
-    private String bootstrapAddress;
-
-    @Value(value = "${message.topic.name}")
-    private String topicName;
-
-    @Value(value = "${long.message.topic.name}")
-    private String longMsgTopicName;
-
-    @Value(value = "${partitioned.topic.name}")
-    private String partitionedTopicName;
-
-    @Value(value = "${filtered.topic.name}")
-    private String filteredTopicName;
-
-    @Value(value = "${greeting.topic.name}")
-    private String greetingTopicName;
-
-    @Value(value = "${multi.type.topic.name}")
-    private String multiTypeTopicName;
-
-    @Bean
-    public KafkaAdmin kafkaAdmin() {
-        Map<String, Object> configs = new HashMap<>();
-        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        return new KafkaAdmin(configs);
-    }
-
-    @Bean
-    public NewTopic topic1() {
-        return new NewTopic(topicName, 1, (short) 1);
-    }
-
-    @Bean
-    public NewTopic topic2() {
-        return new NewTopic(partitionedTopicName, 6, (short) 1);
-    }
-
-    @Bean
-    public NewTopic topic3() {
-        return new NewTopic(filteredTopicName, 1, (short) 1);
-    }
-
-    @Bean
-    public NewTopic topic4() {
-        return new NewTopic(greetingTopicName, 1, (short) 1);
-    }
-
-    @Bean
-    public NewTopic topic5() {
-        NewTopic newTopic = new NewTopic(longMsgTopicName, 1, (short) 1);
-        Map<String, String> configs = new HashMap<>();
-        configs.put("max.message.bytes", "20971520");
-        newTopic.configs(configs);
-        return newTopic;
-    }
-
-    @Bean
-    public NewTopic multiTypeTopic() {
-        return new NewTopic(multiTypeTopicName, 1, (short) 1);
-    }
-}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/MultiTypeKafkaListener.java

@@ -1,32 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-import org.springframework.kafka.annotation.KafkaHandler;
-import org.springframework.kafka.annotation.KafkaListener;
-import org.springframework.messaging.MessagingException;
-import org.springframework.stereotype.Component;
-
-@Component
-@KafkaListener(id = "multiGroup", topics = "multitype")
-public class MultiTypeKafkaListener {
-
-    @KafkaHandler
-    //@RetryableTopic(backoff = @Backoff(value = 3000L), attempts = "5", autoCreateTopics = "false",include = SocketTimeoutException.class, exclude = NullPointerException.class)
-    public void handleGreeting(Greeting greeting) {
-        if (greeting.getName()
-          .equalsIgnoreCase("test")) {
-            throw new MessagingException("test not allowed");
-        }
-        System.out.println("Greeting received: " + greeting);
-    }
-
-    @KafkaHandler
-    public void handleF(Farewell farewell) {
-        System.out.println("Farewell received: " + farewell);
-    }
-
-    @KafkaHandler(isDefault = true)
-    public void unknown(Object object) {
-        System.out.println("Unkown type received: " + object);
-    }
-
-}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/retryable/RetryableApplicationKafkaApp.java

@@ -1,14 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.annotation.Import;
-
-@SpringBootApplication
-@Import(value = { KafkaTopicConfig.class, KafkaConsumerConfig.class, KafkaProducerConfig.class })
-public class RetryableApplicationKafkaApp {
-
-    public static void main(String[] args)  {
-        SpringApplication.run(RetryableApplicationKafkaApp.class, args);
-    }
-}

spring-kafka-2/src/main/resources/application-retry.properties

@@ -1,26 +0,0 @@
-spring.kafka.bootstrap-servers=localhost:9093
-message.topic.name=baeldung
-long.message.topic.name=longMessage
-greeting.topic.name=greeting
-filtered.topic.name=filtered
-partitioned.topic.name=partitioned
-multi.type.topic.name=multitype
-# monitoring - lag analysis
-monitor.kafka.bootstrap.config=localhost:9092
-monitor.kafka.consumer.groupid=baeldungGrp
-monitor.topic.name=baeldung
-# monitoring - simulation
-monitor.producer.simulate=true
-monitor.consumer.simulate=true
-monitor.kafka.consumer.groupid.simulate=baeldungGrpSimulate
-test.topic=testtopic1
-kafka.backoff.interval=9000
-kafka.backoff.max_failure=5
-# multiple listeners properties
-multiple-listeners.books.topic.name=books
-
-spring.kafka.streams.application-id=baeldung-streams
-spring.kafka.consumer.group-id=baeldung-group
-spring.kafka.streams.properties[default.key.serde]=org.apache.kafka.common.serialization.Serdes$StringSerde
-kafka.topics.iot=iot_sensor_data
-

spring-kafka-2/src/test/java/com/baeldung/spring/kafka/retryable/KafkaRetryableIntegrationTest.java

@@ -1,88 +0,0 @@
-package com.baeldung.spring.kafka.retryable;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.jupiter.api.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener;
-import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
-
-import com.baeldung.spring.kafka.retryable.Greeting;
-import com.baeldung.spring.kafka.retryable.RetryableApplicationKafkaApp;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.springframework.test.context.ActiveProfiles;
-
-@SpringBootTest(classes = RetryableApplicationKafkaApp.class)
-@EmbeddedKafka(partitions = 1, controlledShutdown = true, brokerProperties = { "listeners=PLAINTEXT://localhost:9093", "port=9093" })
-@ActiveProfiles("retry")
-public class KafkaRetryableIntegrationTest {
-    @ClassRule
-    public static EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaBroker(1, true, "multitype");
-
-    @Autowired
-    private KafkaListenerEndpointRegistry registry;
-
-    @Autowired
-    private KafkaTemplate<String, String> template;
-
-    private ObjectMapper objectMapper = new ObjectMapper();
-
-    private static final String CONTAINER_GROUP = "multiGroup";
-
-    private static final String TOPIC = "topic";
-
-    @Before
-    public void setup() {
-        System.setProperty("spring.kafka.bootstrap-servers", embeddedKafka.getBrokersAsString());
-    }
-
-    @Test
-    public void givenEmbeddedKafkaBroker_whenSendingAWellFormedMessage_thenMessageIsConsumed() throws Exception {
-        ConcurrentMessageListenerContainer<?, ?> container = (ConcurrentMessageListenerContainer<?, ?>) registry.getListenerContainer(CONTAINER_GROUP);
-        container.stop();
-        @SuppressWarnings("unchecked") AcknowledgingConsumerAwareMessageListener<String, String> messageListener = (AcknowledgingConsumerAwareMessageListener<String, String>) container.getContainerProperties()
-          .getMessageListener();
-        CountDownLatch latch = new CountDownLatch(1);
-        container.getContainerProperties()
-          .setMessageListener((AcknowledgingConsumerAwareMessageListener<String, String>) (data, acknowledgment, consumer) -> {
-              messageListener.onMessage(data, acknowledgment, consumer);
-              latch.countDown();
-          });
-        Greeting greeting = new Greeting("test1", "test2");
-        container.start();
-        template.send(TOPIC, objectMapper.writeValueAsString(greeting));
-        assertThat(latch.await(10, TimeUnit.SECONDS)).isFalse();
-    }
-
-    @Test
-    public void givenEmbeddedKafkaBroker_whenSendingAMalFormedMessage_thenMessageIsConsumedAfterRetry() throws Exception {
-        ConcurrentMessageListenerContainer<?, ?> container = (ConcurrentMessageListenerContainer<?, ?>) registry.getListenerContainer(CONTAINER_GROUP);
-        container.stop();
-        @SuppressWarnings("unchecked") AcknowledgingConsumerAwareMessageListener<String, String> messageListener = (AcknowledgingConsumerAwareMessageListener<String, String>) container.getContainerProperties()
-          .getMessageListener();
-        CountDownLatch latch = new CountDownLatch(1);
-        container.getContainerProperties()
-          .setMessageListener((AcknowledgingConsumerAwareMessageListener<String, String>) (data, acknowledgment, consumer) -> {
-              messageListener.onMessage(data, acknowledgment, consumer);
-              latch.countDown();
-          });
-        container.start();
-        Greeting greeting = new Greeting("test", "test");
-        template.send(TOPIC, objectMapper.writeValueAsString(greeting));
-        //this message will go on error
-        Greeting greeting2 = new Greeting("test2", "test2");
-        template.send(TOPIC, objectMapper.writeValueAsString(greeting2));
-        assertThat(latch.getCount()).isEqualTo(1);
-    }
-
-}