diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index fa380b9e..7c34004e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -470,6 +470,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } // handle the last manual acks, after the listeners have closed handleManualAcks(); + processCommits(); if (this.offsets.size() > 0) { // we always commit after stopping the invoker commitIfNecessary(); @@ -602,6 +603,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } catch (Exception e) { + if (this.containerProperties.isAckOnError()) { + this.acks.add(record); + } if (this.containerProperties.getErrorHandler() != null) { this.containerProperties.getErrorHandler().handle(e, record); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java index 3919234f..411caef8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java @@ -181,6 +181,8 @@ public class ContainerProperties { */ private boolean syncCommits = true; + private boolean ackOnError = true; + private Long idleEventInterval; public ContainerProperties(String... topics) { @@ -385,6 +387,21 @@ public class ContainerProperties { this.idleEventInterval = idleEventInterval; } + /** + * Set whether the container should ack messages that throw exceptions or not. This + * works in conjunction with {@link #ackMode} and is effective only when auto ack is + * false; it is not applicable to manual acks. When this property is set to + * {@code true}, all messages handled will be acked. When set to {@code false}, acks + * will be produced only for successful messages. This allows a component that starts + * throwing exceptions consistently to resume from the last successfully processed + * message. Manual acks will be always be applied. + * @param ackOnError whether the container should acknowledge messages that throw + * exceptions. + */ + public void setAckOnError(boolean ackOnError) { + this.ackOnError = ackOnError; + } + public String[] getTopics() { return this.topics; } @@ -473,4 +490,7 @@ public class ContainerProperties { return this.idleEventInterval; } + public boolean isAckOnError() { + return this.ackOnError; + } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 6fc51179..6ffd6ef1 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -22,6 +22,7 @@ import static org.mockito.Matchers.anyLong; import static org.mockito.Mockito.mock; import java.util.ArrayList; +import java.util.Arrays; import java.util.BitSet; import java.util.Collection; import java.util.List; @@ -65,6 +66,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; * @author Gary Russell * @author Artem Bilan * @author Jerome Mirc + * @author Marius Bogoevici * */ public class ConcurrentMessageListenerContainerTests { @@ -87,9 +89,11 @@ public class ConcurrentMessageListenerContainerTests { private static String topic8 = "testTopic8"; + private static String topic9 = "testTopic9"; + @ClassRule public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, - topic6, topic7, topic8); + topic6, topic7, topic8, topic9); @Test public void testAutoCommit() throws Exception { @@ -608,4 +612,56 @@ public class ConcurrentMessageListenerContainerTests { } + + @Test + public void testAckOnErrorRecord() throws Exception { + logger.info("Start ack on error"); + Map props = KafkaTestUtils.consumerProps("test9", "false", embeddedKafka); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + final CountDownLatch latch = new CountDownLatch(4); + ContainerProperties containerProps = new ContainerProperties(topic9); + containerProps.setMessageListener(new MessageListener() { + + @Override + public void onMessage(ConsumerRecord message) { + logger.info("auto ack on error: " + message); + latch.countDown(); + if (message.value().startsWith("b")) { + throw new RuntimeException(); + } + } + }); + containerProps.setSyncCommits(true); + containerProps.setAckMode(AckMode.RECORD); + containerProps.setAckOnError(false); + ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, + containerProps); + container.setConcurrency(2); + container.setBeanName("testAckOnError"); + container.start(); + ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic9); + template.sendDefault(0, 0, "foo"); + template.sendDefault(1, 0, "bar"); + template.sendDefault(0, 0, "baz"); + template.sendDefault(1, 0, "qux"); + template.flush(); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); + container.stop(); + Consumer consumer = cf.createConsumer(); + consumer.assign(Arrays.asList(new TopicPartition(topic9, 0), new TopicPartition(topic9, 1))); + // this consumer is positioned at 1, the next offset after the successfully + // processed 'foo' + // it has not been updated because 'bar' failed + assertThat(consumer.position(new TopicPartition(topic9, 0))).isEqualTo(1); + // this consumer is positioned at 1, the next offset after the successfully + // processed 'qux' + // it has been updated even 'baz' failed + assertThat(consumer.position(new TopicPartition(topic9, 1))).isEqualTo(2); + logger.info("Stop ack on error"); + } + }