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

Catch exceptions thrown from error handler

Browse Source 
In situations like committing acks while consumer group has already rebalanced, 
and the container error handler is instance of `ConsumerAwareErrorHandler` or `ConsumerAwareBatchErrorHandler`, the error handler will throw an exception 
which would not be caught. 
Then the consumer will be dead, it can not receive messages any more.
This commit is contained in:
realcbb
2018-03-16 21:56:42 +08:00
committed by Artem Bilan
parent 17e255d759
commit 6bd58b2b66
2 changed files with 83 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -92,6 +92,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
* @author Artem Bilan
* @author Loic Talhouarne
* @author Vladimir Tsanev
* @author Chen Binbin
* @author Yang Qiju
* @author Tom van den Berge
*/
@@ -719,11 +720,23 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
break;
}
catch (Exception e) {
if (this.containerProperties.getGenericErrorHandler() != null) {
this.containerProperties.getGenericErrorHandler().handle(e, null);
try {
GenericErrorHandler<?> containerErrorHandler = this.containerProperties.getGenericErrorHandler();
if (containerErrorHandler != null) {
if (containerErrorHandler instanceof ConsumerAwareErrorHandler
|| containerErrorHandler instanceof ConsumerAwareBatchErrorHandler) {
containerErrorHandler.handle(e, null, this.consumer);
}
else {
containerErrorHandler.handle(e, null);
}
}
else {
this.logger.error("Container exception", e);
}
}
else {
this.logger.error("Container exception", e);
catch (Exception ex) {
this.logger.error("Container exception", ex);
}
}
}

67
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -134,11 +134,13 @@ public class KafkaMessageListenerContainerTests {
private static String topic18 = "testTopic18";
private static String topic19 = "testTopic19";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17,
topic18);
topic18, topic19);
@Rule
public TestName testName = new TestName();
@@ -1718,6 +1720,69 @@ public class KafkaMessageListenerContainerTests {
container.stop();
}
@Test
public void testExceptionWhenCommitAfterRebalance() throws Exception {
final CountDownLatch rebalanceLatch = new CountDownLatch(2);
final CountDownLatch consumeLatch = new CountDownLatch(7);
Map<String, Object> props = KafkaTestUtils.consumerProps("test19", "false", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 15000);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic19);
containerProps.setMessageListener((MessageListener<Integer, String>) messages -> {
logger.info("listener: " + messages);
consumeLatch.countDown();
try {
Thread.sleep(3000);
}
catch (InterruptedException e) {
e.printStackTrace();
}
});
containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.BATCH);
containerProps.setPollTimeout(100);
containerProps.setAckOnError(false);
containerProps.setErrorHandler(new SeekToCurrentErrorHandler());
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic19);
containerProps.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
logger.info("rebalance occurred.");
rebalanceLatch.countDown();
}
});
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
container.setBeanName("testContainerException");
container.start();
ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
container.pause();
for (int i = 0; i < 6; i++) {
template.sendDefault(0, 0, "a");
}
template.flush();
container.resume();
// should be rebalanced and consume again
assertThat(rebalanceLatch.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(consumeLatch.await(60, TimeUnit.SECONDS)).isTrue();
container.stop();
}
private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
Consumer<?, ?> consumer = spy(
KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));