From c91d491d100e23e12a448eec3df615bd59f7348c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 13 Nov 2019 11:58:41 -0500 Subject: [PATCH] GH-1309: Fix NPE when no syncCommitTimeout set Resolves https://github.com/spring-projects/spring-kafka/issues/1309 When no commit timeout is provided we update the properties in the local container properties. The parent container's properties must also be updated because the parent is passed into the error handler (e.g. for stopping the entire container when an error occurs). Also, fix a problem where the child container instead of the parent was passed into the batch error handler. --- .../KafkaMessageListenerContainer.java | 39 ++++++++------- .../EnableKafkaIntegrationTests.java | 9 ++++ .../SeekToCurrentBatchErrorHandlerTests.java | 48 ++++++++++++++++++- 3 files changed, 77 insertions(+), 19 deletions(-) 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 7b3e6791..fb372156 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 @@ -132,7 +132,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count private static final boolean MICROMETER_PRESENT = ClassUtils.isPresent( "io.micrometer.core.instrument.MeterRegistry", KafkaMessageListenerContainer.class.getClassLoader()); - private final AbstractMessageListenerContainer container; + private final AbstractMessageListenerContainer thisOrParentContainer; private final TopicPartitionOffset[] topicPartitions; @@ -205,7 +205,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count super(consumerFactory, containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); - this.container = container == null ? this : container; + this.thisOrParentContainer = container == null ? this : container; if (topicPartitions != null) { this.topicPartitions = Arrays.stream(topicPartitions) .map(org.springframework.kafka.support.TopicPartitionInitialOffset::toTPO) @@ -230,7 +230,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count super(consumerFactory, containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); - this.container = container == null ? this : container; + this.thisOrParentContainer = container == null ? this : container; if (topicPartitions != null) { this.topicPartitions = Arrays.copyOf(topicPartitions, topicPartitions.length); } @@ -379,28 +379,28 @@ public class KafkaMessageListenerContainer // NOSONAR line count private void publishIdleContainerEvent(long idleTime, Consumer consumer, boolean paused) { if (getApplicationEventPublisher() != null) { getApplicationEventPublisher().publishEvent(new ListenerContainerIdleEvent(this, - this.container, idleTime, getBeanName(), getAssignedPartitions(), consumer, paused)); + this.thisOrParentContainer, idleTime, getBeanName(), getAssignedPartitions(), consumer, paused)); } } private void publishNonResponsiveConsumerEvent(long timeSinceLastPoll, Consumer consumer) { if (getApplicationEventPublisher() != null) { getApplicationEventPublisher().publishEvent( - new NonResponsiveConsumerEvent(this, this.container, timeSinceLastPoll, + new NonResponsiveConsumerEvent(this, this.thisOrParentContainer, timeSinceLastPoll, getBeanName(), getAssignedPartitions(), consumer)); } } private void publishConsumerPausedEvent(Collection partitions) { if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ConsumerPausedEvent(this, this.container, + getApplicationEventPublisher().publishEvent(new ConsumerPausedEvent(this, this.thisOrParentContainer, Collections.unmodifiableCollection(partitions))); } } private void publishConsumerResumedEvent(Collection partitions) { if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ConsumerResumedEvent(this, this.container, + getApplicationEventPublisher().publishEvent(new ConsumerResumedEvent(this, this.thisOrParentContainer, Collections.unmodifiableCollection(partitions))); } } @@ -409,7 +409,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count try { if (getApplicationEventPublisher() != null) { getApplicationEventPublisher().publishEvent( - new ConsumerStoppingEvent(this, this.container, consumer, getAssignedPartitions())); + new ConsumerStoppingEvent(this, this.thisOrParentContainer, consumer, getAssignedPartitions())); } } catch (Exception e) { @@ -419,32 +419,32 @@ public class KafkaMessageListenerContainer // NOSONAR line count private void publishConsumerStoppedEvent() { if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.container)); + getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.thisOrParentContainer)); } } private void publishConsumerStartingEvent() { this.startLatch.countDown(); if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ConsumerStartingEvent(this, this.container)); + getApplicationEventPublisher().publishEvent(new ConsumerStartingEvent(this, this.thisOrParentContainer)); } } private void publishConsumerStartedEvent() { if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ConsumerStartedEvent(this, this.container)); + getApplicationEventPublisher().publishEvent(new ConsumerStartedEvent(this, this.thisOrParentContainer)); } } private void publishConsumerFailedToStart() { if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ConsumerFailedToStartEvent(this, this.container)); + getApplicationEventPublisher().publishEvent(new ConsumerFailedToStartEvent(this, this.thisOrParentContainer)); } } @Override protected AbstractMessageListenerContainer parentOrThis() { - return this.container; + return this.thisOrParentContainer; } @Override @@ -659,6 +659,11 @@ public class KafkaMessageListenerContainer // NOSONAR line count if (this.containerProperties.getSyncCommitTimeout() == null) { // update the property so we can use it directly from code elsewhere this.containerProperties.setSyncCommitTimeout(this.syncCommitTimeout); + if (KafkaMessageListenerContainer.this.thisOrParentContainer != null) { + KafkaMessageListenerContainer.this.thisOrParentContainer + .getContainerProperties() + .setSyncCommitTimeout(this.syncCommitTimeout); + } } this.maxPollInterval = obtainMaxPollInterval(consumerProperties); this.micrometerHolder = obtainMicrometerHolder(); @@ -1108,11 +1113,11 @@ public class KafkaMessageListenerContainer // NOSONAR line count try { if (!this.isBatchListener && this.errorHandler != null) { this.errorHandler.handle(e, Collections.emptyList(), this.consumer, - KafkaMessageListenerContainer.this.container); + KafkaMessageListenerContainer.this.thisOrParentContainer); } else if (this.isBatchListener && this.batchErrorHandler != null) { this.batchErrorHandler.handle(e, new ConsumerRecords(Collections.emptyMap()), this.consumer, - KafkaMessageListenerContainer.this); + KafkaMessageListenerContainer.this.thisOrParentContainer); } else { this.logger.error(e, "Consumer exception"); @@ -1431,7 +1436,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count private void invokeBatchErrorHandler(final ConsumerRecords records, RuntimeException e) { if (this.batchErrorHandler instanceof ContainerAwareBatchErrorHandler) { this.batchErrorHandler.handle(decorateException(e), records, this.consumer, - KafkaMessageListenerContainer.this.container); + KafkaMessageListenerContainer.this.thisOrParentContainer); } else { this.batchErrorHandler.handle(decorateException(e), records, this.consumer); @@ -1672,7 +1677,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count records.add(iterator.next()); } this.errorHandler.handle(decorateException(e), records, this.consumer, - KafkaMessageListenerContainer.this.container); + KafkaMessageListenerContainer.this.thisOrParentContainer); } else { this.errorHandler.handle(decorateException(e), record, this.consumer); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 2635efe4..3658caca 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -28,6 +28,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import java.lang.reflect.Type; +import java.time.Duration; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -341,8 +342,16 @@ public class EnableKafkaIntegrationTests { MessageListenerContainer listenerContainer = registry.getListenerContainer("manualStart"); assertThat(listenerContainer).isNotNull(); assertThat(listenerContainer.isRunning()).isFalse(); + assertThat(listenerContainer.getContainerProperties().getSyncCommitTimeout()).isNull(); this.registry.start(); assertThat(listenerContainer.isRunning()).isTrue(); + assertThat(((ConcurrentMessageListenerContainer) listenerContainer) + .getContainers() + .get(0) + .getContainerProperties().getSyncCommitTimeout()) + .isEqualTo(Duration.ofSeconds(60)); + assertThat(listenerContainer.getContainerProperties().getSyncCommitTimeout()) + .isEqualTo(Duration.ofSeconds(60)); listenerContainer.stop(); assertThat(KafkaTestUtils.getPropertyValue(listenerContainer, "containerProperties.syncCommits", Boolean.class)) .isFalse(); 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 index 71ab881e..03cb461c 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java @@ -19,6 +19,8 @@ package org.springframework.kafka.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; @@ -36,6 +38,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; @@ -95,7 +98,7 @@ public class SeekToCurrentBatchErrorHandlerTests { */ @SuppressWarnings("unchecked") @Test - public void discardRemainingRecordsFromPollAndSeek() throws Exception { + 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(); @@ -121,7 +124,7 @@ public class SeekToCurrentBatchErrorHandlerTests { } @Test - public void testBackOff() { + void testBackOff() { SeekToCurrentBatchErrorHandler eh = new SeekToCurrentBatchErrorHandler(); eh.setBackOff(new FixedBackOff(10L, 3)); @SuppressWarnings("rawtypes") @@ -136,6 +139,46 @@ public class SeekToCurrentBatchErrorHandlerTests { assertThat(System.currentTimeMillis() - t1).isGreaterThanOrEqualTo(100L); } + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Test + void verifyCorrectContainer() throws InterruptedException { + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + final Consumer consumer = mock(Consumer.class); + AtomicBoolean first = new AtomicBoolean(true); + willAnswer(invocation -> { + if (first.getAndSet(false)) { + throw new IllegalStateException("intentional"); + } + Thread.sleep(50); + return new ConsumerRecords(Collections.emptyMap()); + }).given(consumer).poll(any()); + given(consumerFactory.createConsumer(anyString(), anyString(), anyString(), + eq(KafkaTestUtils.defaultPropertyOverrides()))) + .willReturn(consumer); + ContainerProperties containerProperties = new ContainerProperties("foo"); + containerProperties.setGroupId("grp"); + containerProperties.setMessageListener((BatchMessageListener) record -> { }); + containerProperties.setMissingTopicsFatal(false); + ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(consumerFactory, + containerProperties); + AtomicReference parent = new AtomicReference<>(); + CountDownLatch latch = new CountDownLatch(1); + container.setBatchErrorHandler(new ContainerAwareBatchErrorHandler() { + + @Override + public void handle(Exception thrownException, ConsumerRecords data, Consumer consumer, + MessageListenerContainer container) { + + parent.set(container); + latch.countDown(); + } + }); + container.start(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + container.stop(); + assertThat(parent.get()).isSameAs(container); + } + @Configuration @EnableKafka public static class Config { @@ -233,6 +276,7 @@ public class SeekToCurrentBatchErrorHandlerTests { }); factory.setBatchListener(true); factory.getContainerProperties().setTransactionManager(tm()); + factory.setMissingTopicsFatal(false); return factory; }