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