Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-451: Add Container Stopping Error Handlers

Browse Source 
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.
This commit is contained in:
Gary Russell
2017-11-29 13:28:06 -05:00
committed by Artem Bilan
parent 7cd53119d3
commit d2ce4aba1e
17 changed files with 1797 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

7
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -124,11 +124,12 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
for (int i = 0; i < this.concurrency; i++) {
KafkaMessageListenerContainer<K, V> 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);

48
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java Normal file
View File

@@ -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);
}

52
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java Normal file
View File

@@ -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<ConsumerRecord<?, ?>> 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<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
MessageListenerContainer container);
}

67
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandler.java Normal file
View File

@@ -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);
}
}

68
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerStoppingErrorHandler.java Normal file
View File

@@ -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<ConsumerRecord<?, ?>> 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);
}
}

60
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -91,6 +91,8 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
*/
public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
private final AbstractMessageListenerContainer<K, V> container;
private final ConsumerFactory<K, V> consumerFactory;
private final TopicPartitionInitialOffset[] topicPartitions;
@@ -110,7 +112,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
*/
public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
ContainerProperties containerProperties) {
this(consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null);
this(null, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null);
}
/**
@@ -122,8 +124,35 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
*/
public KafkaMessageListenerContainer(ConsumerFactory<K, V> 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<K, V> container,
ConsumerFactory<K, V> 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<K, V> container,
ConsumerFactory<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> extends AbstractMessageListener
Map<TopicPartition, OffsetAndMetadata> 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<K, V> extends AbstractMessageListener
throw e;
}
try {
if (this.errorHandler instanceof RemainingRecordsErrorHandler) {
if (this.errorHandler instanceof ContainerAwareErrorHandler) {
processCommits();
List<ConsumerRecord<?, ?>> 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);

47
spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandler.java Normal file
View File

@@ -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<TopicPartition, Long> 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);
}
}

7
spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java
View File

@@ -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<ConsumerRecord<?, ?>> records,
Consumer<?, ?> consumer) {
Consumer<?, ?> consumer, MessageListenerContainer container) {
Map<TopicPartition, Long> 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);
}
}

207
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java Normal file
View File

@@ -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<String> 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<TopicPartition, List<ConsumerRecord>> 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;
}
}
}

213
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java Normal file
View File

@@ -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<String> 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<TopicPartition, List<ConsumerRecord>> 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<ConsumerRecord<?, ?>> 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;
}
}
}

227
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java Normal file
View File

@@ -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<String> 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<TopicPartition, List<ConsumerRecord>> 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<ConsumerRecord<?, ?>> 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;
}
}
}

226
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java Normal file
View File

@@ -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<TopicPartition, OffsetAndMetadata> 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<String> 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<TopicPartition, List<ConsumerRecord>> 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);
}
}
}

256
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java Normal file
View File

@@ -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<TopicPartition, OffsetAndMetadata> 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<String> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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);
}
}
}

262
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java Normal file
View File

@@ -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<TopicPartition, OffsetAndMetadata> 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<String> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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);
}
}
}

78
src/reference/asciidoc/kafka.adoc
View File

@@ -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<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> 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<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
...
factory.getContainerProperties().setErrorHandler(myErrorHandler);
...
return factory;
}
----
or
[source, java]
----
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> 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

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<serdes>> 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 <<annotation-error-handling>> for more information.