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

GH-1309: Fix NPE when no syncCommitTimeout set

Browse Source 
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.
This commit is contained in:
Gary Russell
2019-11-13 11:58:41 -05:00
committed by Artem Bilan
parent e4e59f9324
commit c91d491d10
3 changed files with 77 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -132,7 +132,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private static final boolean MICROMETER_PRESENT = ClassUtils.isPresent(
"io.micrometer.core.instrument.MeterRegistry", KafkaMessageListenerContainer.class.getClassLoader());
private final AbstractMessageListenerContainer<K, V> container;
private final AbstractMessageListenerContainer<K, V> thisOrParentContainer;
private final TopicPartitionOffset[] topicPartitions;
@@ -205,7 +205,7 @@ public class KafkaMessageListenerContainer<K, V> // 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<K, V> // 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<K, V> // 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<TopicPartition> 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<TopicPartition> 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<K, V> // 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<K, V> // 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<K, V> // 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<K, V> // 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<K, V>(Collections.emptyMap()), this.consumer,
KafkaMessageListenerContainer.this);
KafkaMessageListenerContainer.this.thisOrParentContainer);
}
else {
this.logger.error(e, "Consumer exception");
@@ -1431,7 +1436,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private void invokeBatchErrorHandler(final ConsumerRecords<K, V> 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<K, V> // 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);

9
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

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

48
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java
View File

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