From d2ce4aba1e68de9e97a63cb0052f7c83e90e40c2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 29 Nov 2017 13:28:06 -0500 Subject: [PATCH] GH-451: Add Container Stopping Error Handlers Resolves https://github.com/spring-projects/spring-kafka/issues/451 Add error handlers that stop the container. Polishing Polishing - Fix SeekToCurrent Error Handers These also have to throw an exception to force a rollback if transactions are enabled. In doing so, I found a bug when using transactions with AckMode.RECORD. Parts of this commit will need to be back ported. --- build.gradle | 2 +- .../ConcurrentMessageListenerContainer.java | 7 +- .../ContainerAwareBatchErrorHandler.java | 48 ++++ .../listener/ContainerAwareErrorHandler.java | 52 ++++ .../ContainerStoppingBatchErrorHandler.java | 67 +++++ .../ContainerStoppingErrorHandler.java | 68 +++++ .../KafkaMessageListenerContainer.java | 60 +++- .../SeekToCurrentBatchErrorHandler.java | 47 ++++ .../listener/SeekToCurrentErrorHandler.java | 7 +- ...ntainerStoppingBatchErrorHandlerTests.java | 207 ++++++++++++++ ...nerStoppingErrorHandlerBatchModeTests.java | 213 ++++++++++++++ ...erStoppingErrorHandlerRecordModeTests.java | 227 +++++++++++++++ .../SeekToCurrentBatchErrorHandlerTests.java | 226 +++++++++++++++ .../SeekToCurrentOnErrorBatchModeTXTests.java | 256 +++++++++++++++++ ...SeekToCurrentOnErrorRecordModeTXTests.java | 262 ++++++++++++++++++ src/reference/asciidoc/kafka.adoc | 78 ++++-- src/reference/asciidoc/whats-new.adoc | 6 + 17 files changed, 1797 insertions(+), 36 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandler.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingErrorHandler.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandler.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java diff --git a/build.gradle b/build.gradle index f4160738..78fca2e7 100644 --- a/build.gradle +++ b/build.gradle @@ -79,7 +79,7 @@ subprojects { subproject -> scalaVersion = '2.11' slf4jVersion = '1.7.25' springRetryVersion = '1.2.1.RELEASE' - springVersion = '5.0.1.RELEASE' + springVersion = '5.0.2.RELEASE' idPrefix = 'kafka' diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index cfdc7514..1d45efe9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -124,11 +124,12 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis for (int i = 0; i < this.concurrency; i++) { KafkaMessageListenerContainer container; if (topicPartitions == null) { - container = new KafkaMessageListenerContainer<>(this.consumerFactory, containerProperties); + container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, + containerProperties); } else { - container = new KafkaMessageListenerContainer<>(this.consumerFactory, containerProperties, - partitionSubset(containerProperties, i)); + container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, + containerProperties, partitionSubset(containerProperties, i)); } if (getBeanName() != null) { container.setBeanName(getBeanName() + "-" + i); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java new file mode 100644 index 00000000..e579e250 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java @@ -0,0 +1,48 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecords; + +/** + * An error handler that has access to the batch of records from the last poll the + * consumer, and the container. + * + * @author Gary Russell + * @since 2.1 + * + */ +@FunctionalInterface +public interface ContainerAwareBatchErrorHandler extends ConsumerAwareBatchErrorHandler { + + @Override + default void handle(Exception thrownException, ConsumerRecords data, Consumer consumer) { + throw new UnsupportedOperationException("Container should never call this"); + } + + /** + * Handle the exception. + * @param thrownException the exception. + * @param data the consumer records. + * @param consumer the consumer. + * @param container the container. + */ + void handle(Exception thrownException, ConsumerRecords data, Consumer consumer, + MessageListenerContainer container); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java new file mode 100644 index 00000000..9939233c --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java @@ -0,0 +1,52 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import java.util.List; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +/** + * An error handler that has access to the unprocessed records from the last poll + * (including the failed record), the consumer, and the container. + * The records passed to the handler will not be passed to the listener + * (unless re-fetched if the handler performs seeks). + * + * @author Gary Russell + * @since 2.1 + * + */ +@FunctionalInterface +public interface ContainerAwareErrorHandler extends RemainingRecordsErrorHandler { + + @Override + default void handle(Exception thrownException, List> records, Consumer consumer) { + throw new UnsupportedOperationException("Container should never call this"); + } + + /** + * Handle the exception. + * @param thrownException the exception. + * @param records the remaining records including the one that failed. + * @param consumer the consumer. + * @param container the container. + */ + void handle(Exception thrownException, List> records, Consumer consumer, + MessageListenerContainer container); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandler.java new file mode 100644 index 00000000..22fa168c --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandler.java @@ -0,0 +1,67 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import java.util.concurrent.Executor; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecords; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.kafka.KafkaException; +import org.springframework.util.Assert; + +/** + * A container error handler that stops the container after an exception + * is thrown by the listener. + * + * @author Gary Russell + * @since 2.1 + * + */ +public class ContainerStoppingBatchErrorHandler implements ContainerAwareBatchErrorHandler { + + private final Executor executor; + + public ContainerStoppingBatchErrorHandler() { + this.executor = new SimpleAsyncTaskExecutor(); + } + + public ContainerStoppingBatchErrorHandler(Executor executor) { + Assert.notNull(executor, "'executor' cannot be null"); + this.executor = executor; + } + + @Override + public void handle(Exception thrownException, ConsumerRecords data, Consumer consumer, + MessageListenerContainer container) { + this.executor.execute(() -> container.stop()); + // isRunning is false before the container.stop() waits for listener thread + int n = 0; + while (container.isRunning() && n++ < 100) { + try { + Thread.sleep(100); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + throw new KafkaException("Stopped container", thrownException); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingErrorHandler.java new file mode 100644 index 00000000..ba148d61 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingErrorHandler.java @@ -0,0 +1,68 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import java.util.List; +import java.util.concurrent.Executor; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.kafka.KafkaException; +import org.springframework.util.Assert; + +/** + * A container error handler that stops the container after an exception + * is thrown by the listener. + * + * @author Gary Russell + * @since 2.1 + * + */ +public class ContainerStoppingErrorHandler implements ContainerAwareErrorHandler { + + private final Executor executor; + + public ContainerStoppingErrorHandler() { + this.executor = new SimpleAsyncTaskExecutor(); + } + + public ContainerStoppingErrorHandler(Executor executor) { + Assert.notNull(executor, "'executor' cannot be null"); + this.executor = executor; + } + + @Override + public void handle(Exception thrownException, List> records, Consumer consumer, + MessageListenerContainer container) { + this.executor.execute(() -> container.stop()); + // isRunning is false before the container.stop() waits for listener thread + int n = 0; + while (container.isRunning() && n++ < 100) { + try { + Thread.sleep(100); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + throw new KafkaException("Stopped container", thrownException); + } + +} 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 41926667..6bf94c9a 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 @@ -91,6 +91,8 @@ import org.springframework.util.concurrent.ListenableFutureCallback; */ public class KafkaMessageListenerContainer extends AbstractMessageListenerContainer { + private final AbstractMessageListenerContainer container; + private final ConsumerFactory consumerFactory; private final TopicPartitionInitialOffset[] topicPartitions; @@ -110,7 +112,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener */ public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { - this(consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null); + this(null, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null); } /** @@ -122,8 +124,35 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener */ public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) { + this(null, consumerFactory, containerProperties, topicPartitions); + } + + /** + * Construct an instance with the supplied configuration properties. + * @param container a delegating container (if this is a sub-container). + * @param consumerFactory the consumer factory. + * @param containerProperties the container properties. + */ + KafkaMessageListenerContainer(AbstractMessageListenerContainer container, + ConsumerFactory consumerFactory, + ContainerProperties containerProperties) { + this(container, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null); + } + + /** + * Construct an instance with the supplied configuration properties and specific + * topics/partitions/initialOffsets. + * @param container a delegating container (if this is a sub-container). + * @param consumerFactory the consumer factory. + * @param containerProperties the container properties. + * @param topicPartitions the topics/partitions; duplicates are eliminated. + */ + KafkaMessageListenerContainer(AbstractMessageListenerContainer container, + ConsumerFactory consumerFactory, + ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) { super(containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); + this.container = container == null ? this : container; this.consumerFactory = consumerFactory; if (topicPartitions != null) { this.topicPartitions = Arrays.copyOf(topicPartitions, topicPartitions.length); @@ -408,8 +437,6 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Assert.state(!this.isBatchListener || !this.isRecordAck, "Cannot use AckMode.RECORD with a batch listener"); if (this.transactionManager != null) { this.transactionTemplate = new TransactionTemplate(this.transactionManager); - Assert.state(!(this.errorHandler instanceof RemainingRecordsErrorHandler), - "You cannot use a 'RemainingRecordsErrorHandler' with transactions"); } else { this.transactionTemplate = null; @@ -833,7 +860,13 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener throw e; } try { - this.batchErrorHandler.handle(e, records, this.consumer); + if (this.batchErrorHandler instanceof ContainerAwareBatchErrorHandler) { + ((ContainerAwareBatchErrorHandler) this.batchErrorHandler) + .handle(e, records, this.consumer, KafkaMessageListenerContainer.this.container); + } + else { + this.batchErrorHandler.handle(e, records, this.consumer); + } if (producer != null) { sendOffsetsToTransaction(producer); } @@ -884,7 +917,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener producer = ((KafkaResourceHolder) TransactionSynchronizationManager .getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())).getProducer(); } - RuntimeException aborted = doInvokeRecordListener(record, producer, null); + RuntimeException aborted = doInvokeRecordListener(record, producer, iterator); if (aborted != null) { throw aborted; } @@ -949,12 +982,16 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Map offsetsToCommit = Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)); - - if (this.containerProperties.isSyncCommits()) { - this.consumer.commitSync(offsetsToCommit); + if (producer == null) { + if (this.containerProperties.isSyncCommits()) { + this.consumer.commitSync(offsetsToCommit); + } + else { + this.consumer.commitAsync(offsetsToCommit, this.commitCallback); + } } else { - this.consumer.commitAsync(offsetsToCommit, this.commitCallback); + this.acks.add(record); } } else if (!this.isAnyManualAck && !this.autoCommit) { @@ -972,14 +1009,15 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener throw e; } try { - if (this.errorHandler instanceof RemainingRecordsErrorHandler) { + if (this.errorHandler instanceof ContainerAwareErrorHandler) { processCommits(); List> records = new ArrayList<>(); records.add(record); while (iterator.hasNext()) { records.add(iterator.next()); } - ((RemainingRecordsErrorHandler) this.errorHandler).handle(e, records, this.consumer); + ((ContainerAwareErrorHandler) this.errorHandler).handle(e, records, this.consumer, + KafkaMessageListenerContainer.this.container); } else { this.errorHandler.handle(e, record, this.consumer); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandler.java new file mode 100644 index 00000000..c328e8db --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandler.java @@ -0,0 +1,47 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; + +import org.springframework.kafka.KafkaException; + +/** + * An error handler that seeks to the current offset for each topic in batch of records. + * Used to rewind partitions after a message failure so that the batch can be replayed. + * + * @author Gary Russell + * @since 2.1 + * + */ +public class SeekToCurrentBatchErrorHandler implements ContainerAwareBatchErrorHandler { + + @Override + public void handle(Exception thrownException, ConsumerRecords data, Consumer consumer, + MessageListenerContainer container) { + Map offsets = new LinkedHashMap<>(); + data.forEach(r -> offsets.computeIfAbsent(new TopicPartition(r.topic(), r.partition()), k -> r.offset())); + offsets.forEach(consumer::seek); + throw new KafkaException("Seek to current after exception", thrownException); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java index c013ebea..a822de39 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java @@ -24,6 +24,8 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.TopicPartition; +import org.springframework.kafka.KafkaException; + /** * An error handler that seeks to the current offset for each topic in the remaining * records. Used to rewind partitions after a message failure so that it can be @@ -33,15 +35,16 @@ import org.apache.kafka.common.TopicPartition; * @since 2.0.1 * */ -public class SeekToCurrentErrorHandler implements RemainingRecordsErrorHandler { +public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler { @Override public void handle(Exception thrownException, List> records, - Consumer consumer) { + Consumer consumer, MessageListenerContainer container) { Map offsets = new LinkedHashMap<>(); records.forEach(r -> offsets.computeIfAbsent(new TopicPartition(r.topic(), r.partition()), k -> r.offset())); offsets.forEach(consumer::seek); + throw new KafkaException("Seek to current after exception", thrownException); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java new file mode 100644 index 00000000..6f170e5e --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java @@ -0,0 +1,207 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.InOrder; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.1 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class ContainerStoppingBatchErrorHandlerTests { + + private static final String CONTAINER_ID = "container"; + + @SuppressWarnings("rawtypes") + @Autowired + private Consumer consumer; + + @Autowired + private Config config; + + @Autowired + private KafkaListenerEndpointRegistry registry; + + /* + * Deliver 6 records from three partitions, fail on the second record second + * partition. + */ + @SuppressWarnings("unchecked") + @Test + public void stopContainerAfterException() throws Exception { + assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID); + assertThat(container.isRunning()).isFalse(); + InOrder inOrder = inOrder(this.consumer); + inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.consumer).wakeup(); + inOrder.verify(this.consumer).unsubscribe(); + inOrder.verify(this.consumer).close(); + inOrder.verifyNoMoreInteractions(); + } + + @Configuration + @EnableKafka + public static class Config { + + private final CountDownLatch pollLatch = new CountDownLatch(1); + + private final CountDownLatch deliveryLatch = new CountDownLatch(1); + + private final CountDownLatch errorLatch = new CountDownLatch(1); + + private final CountDownLatch closeLatch = new CountDownLatch(1); + + private final CountDownLatch commitLatch = new CountDownLatch(3); + + @KafkaListener(id = CONTAINER_ID, topics = "foo") + public void foo(List in) { + this.deliveryLatch.countDown(); + throw new RuntimeException("foo"); + } + + @SuppressWarnings({ "rawtypes" }) + @Bean + public ConsumerFactory consumerFactory() { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = consumer(); + given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer); + return consumerFactory; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public Consumer consumer() { + final Consumer consumer = mock(Consumer.class); + final TopicPartition topicPartition0 = new TopicPartition("foo", 0); + final TopicPartition topicPartition1 = new TopicPartition("foo", 1); + final TopicPartition topicPartition2 = new TopicPartition("foo", 2); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)).onPartitionsAssigned( + Collections.singletonList(topicPartition1)); + return null; + }).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition0, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + records1.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + records1.put(topicPartition2, Arrays.asList( + new ConsumerRecord("foo", 2, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"), + new ConsumerRecord("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + final AtomicInteger which = new AtomicInteger(); + willAnswer(i -> { + this.pollLatch.countDown(); + switch (which.getAndIncrement()) { + case 0: + return new ConsumerRecords(records1); + default: + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new ConsumerRecords(Collections.emptyMap()); + } + }).given(consumer).poll(1000); + willAnswer(i -> { + this.commitLatch.countDown(); + return null; + }).given(consumer).commitSync(any(Map.class)); + willAnswer(i -> { + this.closeLatch.countDown(); + return null; + }).given(consumer).close(); + return consumer; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setAckOnError(false); + factory.getContainerProperties().setBatchErrorHandler(new ContainerStoppingBatchErrorHandler() { + + @Override + public void handle(Exception thrownException, ConsumerRecords records, + Consumer consumer, MessageListenerContainer container) { + RuntimeException exception = null; + try { + super.handle(thrownException, records, consumer, container); + } + catch (RuntimeException e) { + exception = e; + } + errorLatch.countDown(); + throw exception; + } + + }); + factory.setBatchListener(true); + return factory; + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java new file mode 100644 index 00000000..e4601091 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java @@ -0,0 +1,213 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.InOrder; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.1 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class ContainerStoppingErrorHandlerBatchModeTests { + + private static final String CONTAINER_ID = "container"; + + @SuppressWarnings("rawtypes") + @Autowired + private Consumer consumer; + + @Autowired + private Config config; + + @Autowired + private KafkaListenerEndpointRegistry registry; + + /* + * Deliver 6 records from three partitions, fail on the second record second + * partition. + */ + @SuppressWarnings("unchecked") + @Test + public void stopContainerAfterException() throws Exception { + assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID); + assertThat(container.isRunning()).isFalse(); + InOrder inOrder = inOrder(this.consumer); + inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.consumer).wakeup(); + inOrder.verify(this.consumer).unsubscribe(); + inOrder.verify(this.consumer).close(); + inOrder.verifyNoMoreInteractions(); + assertThat(this.config.count).isEqualTo(4); + assertThat(this.config.contents.toArray()).isEqualTo(new String[] + { "foo", "bar", "baz", "qux" }); + } + + @Configuration + @EnableKafka + public static class Config { + + private final List contents = new ArrayList<>(); + + private final CountDownLatch pollLatch = new CountDownLatch(1); + + private final CountDownLatch deliveryLatch = new CountDownLatch(3); + + private final CountDownLatch errorLatch = new CountDownLatch(1); + + private final CountDownLatch closeLatch = new CountDownLatch(1); + + private int count; + + @KafkaListener(id = CONTAINER_ID, topics = "foo") + public void foo(String in) { + this.contents.add(in); + this.deliveryLatch.countDown(); + if (++this.count == 4) { // part 1, offset 1, first time + throw new RuntimeException("foo"); + } + } + + @SuppressWarnings({ "rawtypes" }) + @Bean + public ConsumerFactory consumerFactory() { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = consumer(); + given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer); + return consumerFactory; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public Consumer consumer() { + final Consumer consumer = mock(Consumer.class); + final TopicPartition topicPartition0 = new TopicPartition("foo", 0); + final TopicPartition topicPartition1 = new TopicPartition("foo", 1); + final TopicPartition topicPartition2 = new TopicPartition("foo", 2); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)).onPartitionsAssigned( + Collections.singletonList(topicPartition1)); + return null; + }).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition0, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + records1.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + records1.put(topicPartition2, Arrays.asList( + new ConsumerRecord("foo", 2, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"), + new ConsumerRecord("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + final AtomicInteger which = new AtomicInteger(); + willAnswer(i -> { + this.pollLatch.countDown(); + switch (which.getAndIncrement()) { + case 0: + return new ConsumerRecords(records1); + default: + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new ConsumerRecords(Collections.emptyMap()); + } + }).given(consumer).poll(1000); + willAnswer(i -> { + this.closeLatch.countDown(); + return null; + }).given(consumer).close(); + return consumer; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setAckOnError(false); + factory.getContainerProperties().setErrorHandler(new ContainerStoppingErrorHandler() { + + @Override + public void handle(Exception thrownException, List> records, + Consumer consumer, MessageListenerContainer container) { + RuntimeException exception = null; + try { + super.handle(thrownException, records, consumer, container); + } + catch (RuntimeException e) { + exception = e; + } + errorLatch.countDown(); + throw exception; + } + + }); + factory.getContainerProperties().setAckMode(AckMode.BATCH); + return factory; + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java new file mode 100644 index 00000000..a29353cd --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java @@ -0,0 +1,227 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.InOrder; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.1 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class ContainerStoppingErrorHandlerRecordModeTests { + + private static final String CONTAINER_ID = "container"; + + @SuppressWarnings("rawtypes") + @Autowired + private Consumer consumer; + + @Autowired + private Config config; + + @Autowired + private KafkaListenerEndpointRegistry registry; + + /* + * Deliver 6 records from three partitions, fail on the second record second + * partition. + */ + @SuppressWarnings("unchecked") + @Test + public void stopContainerAfterException() throws Exception { + assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID); + assertThat(container.isRunning()).isFalse(); + InOrder inOrder = inOrder(this.consumer); + inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.consumer).commitSync( + Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(1L))); + inOrder.verify(this.consumer).commitSync( + Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(2L))); + inOrder.verify(this.consumer).commitSync( + Collections.singletonMap(new TopicPartition("foo", 1), new OffsetAndMetadata(1L))); + inOrder.verify(this.consumer).wakeup(); + inOrder.verify(this.consumer).unsubscribe(); + inOrder.verify(this.consumer).close(); + inOrder.verifyNoMoreInteractions(); + assertThat(this.config.count).isEqualTo(4); + assertThat(this.config.contents.toArray()).isEqualTo(new String[] + { "foo", "bar", "baz", "qux" }); + } + + @Configuration + @EnableKafka + public static class Config { + + private final List contents = new ArrayList<>(); + + private final CountDownLatch pollLatch = new CountDownLatch(1); + + private final CountDownLatch deliveryLatch = new CountDownLatch(3); + + private final CountDownLatch errorLatch = new CountDownLatch(1); + + private final CountDownLatch closeLatch = new CountDownLatch(1); + + private final CountDownLatch commitLatch = new CountDownLatch(3); + + private int count; + + @KafkaListener(id = CONTAINER_ID, topics = "foo") + public void foo(String in) { + this.contents.add(in); + this.deliveryLatch.countDown(); + if (++this.count == 4) { // part 1, offset 1, first time + throw new RuntimeException("foo"); + } + } + + @SuppressWarnings({ "rawtypes" }) + @Bean + public ConsumerFactory consumerFactory() { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = consumer(); + given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer); + return consumerFactory; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public Consumer consumer() { + final Consumer consumer = mock(Consumer.class); + final TopicPartition topicPartition0 = new TopicPartition("foo", 0); + final TopicPartition topicPartition1 = new TopicPartition("foo", 1); + final TopicPartition topicPartition2 = new TopicPartition("foo", 2); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)).onPartitionsAssigned( + Collections.singletonList(topicPartition1)); + return null; + }).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition0, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + records1.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + records1.put(topicPartition2, Arrays.asList( + new ConsumerRecord("foo", 2, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"), + new ConsumerRecord("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + final AtomicInteger which = new AtomicInteger(); + willAnswer(i -> { + this.pollLatch.countDown(); + switch (which.getAndIncrement()) { + case 0: + return new ConsumerRecords(records1); + default: + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new ConsumerRecords(Collections.emptyMap()); + } + }).given(consumer).poll(1000); + willAnswer(i -> { + this.commitLatch.countDown(); + return null; + }).given(consumer).commitSync(any(Map.class)); + willAnswer(i -> { + this.closeLatch.countDown(); + return null; + }).given(consumer).close(); + return consumer; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setAckOnError(false); + factory.getContainerProperties().setErrorHandler(new ContainerStoppingErrorHandler() { + + @Override + public void handle(Exception thrownException, List> records, + Consumer consumer, MessageListenerContainer container) { + RuntimeException exception = null; + try { + super.handle(thrownException, records, consumer, container); + } + catch (RuntimeException e) { + exception = e; + } + errorLatch.countDown(); + throw exception; + } + + }); + factory.getContainerProperties().setAckMode(AckMode.RECORD); + return factory; + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java new file mode 100644 index 00000000..3e8408ae --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java @@ -0,0 +1,226 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.InOrder; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.1 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class SeekToCurrentBatchErrorHandlerTests { + + private static final String CONTAINER_ID = "container"; + + @SuppressWarnings("rawtypes") + @Autowired + private Consumer consumer; + + @SuppressWarnings("rawtypes") + @Autowired + private Producer producer; + + @Autowired + private Config config; + + @Autowired + private KafkaListenerEndpointRegistry registry; + + /* + * Deliver 6 records from three partitions, fail on the second record second + * partition. + */ + @SuppressWarnings("unchecked") + @Test + public void discardRemainingRecordsFromPollAndSeek() throws Exception { + assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue(); + this.registry.stop(); + assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + InOrder inOrder = inOrder(this.consumer, this.producer); + inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.producer).beginTransaction(); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 0), 0L); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 0L); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L); + inOrder.verify(this.producer).abortTransaction(); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.producer).beginTransaction(); + Map offsets = new LinkedHashMap<>(); + offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L)); + offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L)); + offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + } + + @Configuration + @EnableKafka + public static class Config { + + private final CountDownLatch pollLatch = new CountDownLatch(1); + + private final CountDownLatch deliveryLatch = new CountDownLatch(2); + + private final CountDownLatch closeLatch = new CountDownLatch(1); + + private final AtomicBoolean fail = new AtomicBoolean(true); + + @KafkaListener(id = CONTAINER_ID, topics = "foo") + public void foo(List in) { + this.deliveryLatch.countDown(); + if (this.fail.getAndSet(false)) { + throw new RuntimeException("foo"); + } + } + + @SuppressWarnings({ "rawtypes" }) + @Bean + public ConsumerFactory consumerFactory() { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = consumer(); + given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer); + return consumerFactory; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public Consumer consumer() { + final Consumer consumer = mock(Consumer.class); + final TopicPartition topicPartition0 = new TopicPartition("foo", 0); + final TopicPartition topicPartition1 = new TopicPartition("foo", 1); + final TopicPartition topicPartition2 = new TopicPartition("foo", 2); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)).onPartitionsAssigned( + Collections.singletonList(topicPartition1)); + return null; + }).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition0, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + records1.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + records1.put(topicPartition2, Arrays.asList( + new ConsumerRecord("foo", 2, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"), + new ConsumerRecord("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + final AtomicInteger which = new AtomicInteger(); + willAnswer(i -> { + this.pollLatch.countDown(); + switch (which.getAndIncrement()) { + case 0: + case 1: + return new ConsumerRecords(records1); + default: + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new ConsumerRecords(Collections.emptyMap()); + } + }).given(consumer).poll(1000); + willAnswer(i -> { + this.closeLatch.countDown(); + return null; + }).given(consumer).close(); + return consumer; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setAckOnError(false); + factory.getContainerProperties().setBatchErrorHandler(new SeekToCurrentBatchErrorHandler()); + factory.setBatchListener(true); + factory.getContainerProperties().setTransactionManager(tm()); + return factory; + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Bean + public KafkaTransactionManager tm() { + return new KafkaTransactionManager<>(producerFactory()); + } + + @SuppressWarnings("rawtypes") + @Bean + public ProducerFactory producerFactory() { + ProducerFactory pf = mock(ProducerFactory.class); + given(pf.createProducer()).willReturn(producer()); + given(pf.transactionCapable()).willReturn(true); + return pf; + } + + @SuppressWarnings("rawtypes") + @Bean + public Producer producer() { + return mock(Producer.class); + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java new file mode 100644 index 00000000..909e9ef4 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java @@ -0,0 +1,256 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.InOrder; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; +import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.0.1 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class SeekToCurrentOnErrorBatchModeTXTests { + + private static final String CONTAINER_ID = "container"; + + @SuppressWarnings("rawtypes") + @Autowired + private Consumer consumer; + + @SuppressWarnings("rawtypes") + @Autowired + private Producer producer; + + @Autowired + private Config config; + + @Autowired + private KafkaListenerEndpointRegistry registry; + + /* + * Deliver 6 records from three partitions, fail on the second record second + * partition, first attempt; verify partition 0,1 committed and a total of 7 records + * handled after seek. + */ + @SuppressWarnings("unchecked") + @Test + public void discardRemainingRecordsFromPollAndSeek() throws Exception { + assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue(); + this.registry.stop(); + assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + InOrder inOrder = inOrder(this.consumer, this.producer); + inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.producer).beginTransaction(); + Map offsets = new LinkedHashMap<>(); + offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(1L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(1L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 1L); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L); + inOrder.verify(this.producer).abortTransaction(); + inOrder.verify(this.consumer).poll(1000); + offsets.clear(); + offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(1L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + inOrder.verify(this.consumer).poll(1000); + assertThat(this.config.count).isEqualTo(7); + assertThat(this.config.contents.toArray()).isEqualTo(new String[] + { "foo", "bar", "baz", "qux", "qux", "fiz", "buz" }); + } + + @Configuration + @EnableKafka + public static class Config { + + private final List contents = new ArrayList<>(); + + private final CountDownLatch pollLatch = new CountDownLatch(3); + + private final CountDownLatch deliveryLatch = new CountDownLatch(7); + + private final CountDownLatch closeLatch = new CountDownLatch(1); + + private int count; + + @KafkaListener(id = CONTAINER_ID, topics = "foo") + public void foo(String in) { + this.contents.add(in); + this.deliveryLatch.countDown(); + if (++this.count == 4) { // part 1, offset 1, first time + throw new RuntimeException("foo"); + } + } + + @SuppressWarnings({ "rawtypes" }) + @Bean + public ConsumerFactory consumerFactory() { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = consumer(); + given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer); + return consumerFactory; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public Consumer consumer() { + final Consumer consumer = mock(Consumer.class); + final TopicPartition topicPartition0 = new TopicPartition("foo", 0); + final TopicPartition topicPartition1 = new TopicPartition("foo", 1); + final TopicPartition topicPartition2 = new TopicPartition("foo", 2); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)).onPartitionsAssigned( + Collections.singletonList(topicPartition1)); + return null; + }).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition0, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + records1.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + records1.put(topicPartition2, Arrays.asList( + new ConsumerRecord("foo", 2, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"), + new ConsumerRecord("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + Map> records2 = new LinkedHashMap<>(records1); + records2.remove(topicPartition0); + records2.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + final AtomicInteger which = new AtomicInteger(); + willAnswer(i -> { + this.pollLatch.countDown(); + switch (which.getAndIncrement()) { + case 0: + return new ConsumerRecords(records1); + case 1: + return new ConsumerRecords(records2); + default: + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new ConsumerRecords(Collections.emptyMap()); + } + }).given(consumer).poll(1000); + willAnswer(i -> { + this.closeLatch.countDown(); + return null; + }).given(consumer).close(); + return consumer; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setAckOnError(false); + factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler()); + factory.getContainerProperties().setAckMode(AckMode.BATCH); + factory.getContainerProperties().setTransactionManager(tm()); + return factory; + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Bean + public KafkaTransactionManager tm() { + return new KafkaTransactionManager<>(producerFactory()); + } + + @SuppressWarnings("rawtypes") + @Bean + public ProducerFactory producerFactory() { + ProducerFactory pf = mock(ProducerFactory.class); + given(pf.createProducer()).willReturn(producer()); + given(pf.transactionCapable()).willReturn(true); + return pf; + } + + @SuppressWarnings("rawtypes") + @Bean + public Producer producer() { + return mock(Producer.class); + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java new file mode 100644 index 00000000..f341fab7 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java @@ -0,0 +1,262 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.InOrder; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; +import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.0.1 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class SeekToCurrentOnErrorRecordModeTXTests { + + private static final String CONTAINER_ID = "container"; + + @SuppressWarnings("rawtypes") + @Autowired + private Consumer consumer; + + @SuppressWarnings("rawtypes") + @Autowired + private Producer producer; + + @Autowired + private Config config; + + @Autowired + private KafkaListenerEndpointRegistry registry; + + /* + * Deliver 6 records from three partitions, fail on the second record second + * partition, first attempt; verify partition 0,1 committed and a total of 7 records + * handled after seek. + */ + @SuppressWarnings("unchecked") + @Test + public void discardRemainingRecordsFromPollAndSeek() throws Exception { + assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue(); + this.registry.stop(); + assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + InOrder inOrder = inOrder(this.consumer, this.producer); + inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + inOrder.verify(this.consumer).poll(1000); + inOrder.verify(this.producer).beginTransaction(); + Map offsets = new LinkedHashMap<>(); + offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(1L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(1L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 1L); + inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L); + inOrder.verify(this.producer).abortTransaction(); + inOrder.verify(this.consumer).poll(1000); + offsets.clear(); + offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(1L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + offsets.clear(); + offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L)); + inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID); + inOrder.verify(this.producer).commitTransaction(); + inOrder.verify(this.consumer).poll(1000); + assertThat(this.config.count).isEqualTo(7); + assertThat(this.config.contents.toArray()).isEqualTo(new String[] + { "foo", "bar", "baz", "qux", "qux", "fiz", "buz" }); + } + + @Configuration + @EnableKafka + public static class Config { + + private final List contents = new ArrayList<>(); + + private final CountDownLatch pollLatch = new CountDownLatch(3); + + private final CountDownLatch deliveryLatch = new CountDownLatch(7); + + private final CountDownLatch closeLatch = new CountDownLatch(1); + + private final CountDownLatch commitLatch = new CountDownLatch(7); + + private int count; + + @KafkaListener(id = CONTAINER_ID, topics = "foo") + public void foo(String in) { + this.contents.add(in); + this.deliveryLatch.countDown(); + if (++this.count == 4) { // part 1, offset 1, first time + throw new RuntimeException("foo"); + } + } + + @SuppressWarnings({ "rawtypes" }) + @Bean + public ConsumerFactory consumerFactory() { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = consumer(); + given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer); + return consumerFactory; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public Consumer consumer() { + final Consumer consumer = mock(Consumer.class); + final TopicPartition topicPartition0 = new TopicPartition("foo", 0); + final TopicPartition topicPartition1 = new TopicPartition("foo", 1); + final TopicPartition topicPartition2 = new TopicPartition("foo", 2); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)).onPartitionsAssigned( + Collections.singletonList(topicPartition1)); + return null; + }).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class)); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition0, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + records1.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + records1.put(topicPartition2, Arrays.asList( + new ConsumerRecord("foo", 2, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"), + new ConsumerRecord("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + Map> records2 = new LinkedHashMap<>(records1); + records2.remove(topicPartition0); + records2.put(topicPartition1, Arrays.asList( + new ConsumerRecord("foo", 1, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + final AtomicInteger which = new AtomicInteger(); + willAnswer(i -> { + this.pollLatch.countDown(); + switch (which.getAndIncrement()) { + case 0: + return new ConsumerRecords(records1); + case 1: + return new ConsumerRecords(records2); + default: + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new ConsumerRecords(Collections.emptyMap()); + } + }).given(consumer).poll(1000); + willAnswer(i -> { + this.commitLatch.countDown(); + return null; + }).given(consumer).commitSync(any(Map.class)); + willAnswer(i -> { + this.closeLatch.countDown(); + return null; + }).given(consumer).close(); + return consumer; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setAckOnError(false); + factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler()); + factory.getContainerProperties().setAckMode(AckMode.RECORD); + factory.getContainerProperties().setTransactionManager(tm()); + return factory; + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Bean + public KafkaTransactionManager tm() { + return new KafkaTransactionManager<>(producerFactory()); + } + + @SuppressWarnings("rawtypes") + @Bean + public ProducerFactory producerFactory() { + ProducerFactory pf = mock(ProducerFactory.class); + given(pf.createProducer()).willReturn(producer()); + given(pf.transactionCapable()).willReturn(true); + return pf; + } + + @SuppressWarnings("rawtypes") + @Bean + public Producer producer() { + return mock(Producer.class); + } + + } + +} diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 5df0c057..1dc16926 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1405,24 +1405,7 @@ static class MultiListenerBean { [[annotation-error-handling]] ==== Handling Exceptions -You can specify a global error handler used for all listeners in the container factory. - -[source, java] ----- -@Bean -public KafkaListenerContainerFactory> - kafkaListenerContainerFactory() { - ConcurrentKafkaListenerContainerFactory factory = - new ConcurrentKafkaListenerContainerFactory<>(); - ... - factory.getContainerProperties().setErrorHandler(myErrorHandler); - ... - return factory; -} ----- - -By default, if an annotated listener method throws an exception, it is thrown to the container, and the message will be handled according to the container configuration. -Nothing is returned to the sender. +===== Listener Error Handlers Starting with _version 2.0_, the `@KafkaListener` annotation has a new attribute: `errorHandler`. @@ -1496,7 +1479,45 @@ public ConsumerAwareListenerErrorHandler listen10ErrorHandler() { This resets each topic/partition in the batch to the lowest offset in the batch. -Similarly, the container-level error handler (`ErrorHandler` and `BatchErrorHandler`) have sub-interfaces `ConsumerAwareErrorHandler` and `ConsumerAwareBatchErrorHandler` with method signatures: +===== Container Error Handlers + +You can specify a global error handler used for all listeners in the container factory. + +[source, java] +---- +@Bean +public KafkaListenerContainerFactory> + kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + ... + factory.getContainerProperties().setErrorHandler(myErrorHandler); + ... + return factory; +} +---- + +or + +[source, java] +---- +@Bean +public KafkaListenerContainerFactory> + kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + ... + factory.getContainerProperties().setBatchErrorHandler(myBatchErrorHandler); + ... + return factory; +} +---- + +By default, if an annotated listener method throws an exception, it is thrown to the container, and the message will be handled according to the container configuration. + +===== Consumer-Aware Container Error Handlers + +The container-level error handlers (`ErrorHandler` and `BatchErrorHandler`) have sub-interfaces `ConsumerAwareErrorHandler` and `ConsumerAwareBatchErrorHandler` with method signatures: [source, java] ---- @@ -1511,6 +1532,8 @@ Similar to the `@KafkaListener` error handlers, you can reset the offsets as nee NOTE: Unlike the listener-level error handlers, however, you should set the container property `ackOnError` to false when making adjustments; otherwise any pending acks will be applied after your repositioning. +===== Seek To Current Container Error Handlers + If an `ErrorHandler` implements `RemainingRecordsErrorHandler`, the error handler is provided with the failed record and any unprocessed records retrieved by the previous `poll()`. Those records will not be passed to the listener after the handler exits. @@ -1552,6 +1575,23 @@ The next `poll()` will return the 3 unprocessed records. If the `AckMode` was `BATCH`, the container commits the offsets for the first 2 partitions before calling the error handler. +The `SeekToCurrentBatchErrorHandler` seeks each partition to the first record in each partition in the batch so the whole batch is replayed. + +After seeking, an exception wrapping the `ListenerExecutionFailedException` is thrown. +This is to cause the transaction to roll back (if transactions are enabled). + +===== Container Stopping Error Handlers + +The `ContainerStoppingErrorHandler` (used with record listeners) will stop the container if the listener throws an exception. +When the `AckMode` is `RECORD`, offsets for already processed records will be committed. +When the `AckMode` is any manual, offsets for already acknowledged records will be committed. +When the `AckMode` is `BATCH`, the entire batch will be replayed when the container is restarted, unless transactions are enabled in which case only the unprocessed records will be re-fetched. + +The `ContainerStoppingBatchErrorHandler` (used with batch listeners) will stop the container and the entire batch will be replayed when the container is restarted. + +After the container stops, an exception wrapping the `ListenerExecutionFailedException` is thrown. +This is to cause the transaction to roll back (if transactions are enabled). + [[kerberos]] ==== Kerberos diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 9d8d3cf5..3414ea87 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -8,3 +8,9 @@ This version requires the 1.0.0 `kafka-clients` or higher. The `StringJsonMessageConverter` and `JsonSerializer` now add type information in `Headers`, allowing the converter and `JsonDeserializer` to create specific types on reception, based on the message itself rather than a fixed configured type. See <> for more information. + + +==== Container Stopping Error Handlers + +Container Error handlers are now provided for both record and batch listeners that treat any exceptions thrown by the listener as fatal; they stop the container. +See <> for more information.