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

GH-550: master to 2.2; fix tangles

Browse Source 
Fixes https://github.com/spring-projects/spring-kafka/issues/550

- package tangle `....listener` and `....listener.config` - remove config package
- class tangles between `ContainerProperties` and the listener containers
  - AckMode moved to properties
  - Error handler setters moved from properties to containers
This commit is contained in:
Gary Russell
2018-04-03 16:07:26 -04:00
parent e6985e813c
commit b048aaa8f0
29 changed files with 292 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
gradle.properties
View File

@@ -1 +1 @@
version=2.1.6.BUILD-SNAPSHOT
version=2.2.0.BUILD-SNAPSHOT

30
spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
View File

@@ -26,9 +26,10 @@ import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.BatchErrorHandler;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.GenericErrorHandler;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
@@ -51,6 +52,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null);
private GenericErrorHandler<?> errorHandler;
private ConsumerFactory<K, V> consumerFactory;
private Boolean autoStartup;
@@ -191,6 +194,24 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
this.replyTemplate = replyTemplate;
}
/**
* Set the error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
* @since 2.2
*/
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Set the batch error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
* @since 2.2
*/
public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Obtain the properties template for this factory - set properties as needed
* and they will be copied to a final properties instance for the endpoint.
@@ -274,11 +295,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
if (this.containerProperties.getAckTime() > 0) {
properties.setAckTime(this.containerProperties.getAckTime());
}
if (this.containerProperties.getGenericErrorHandler() instanceof BatchErrorHandler) {
properties.setBatchErrorHandler((BatchErrorHandler) this.containerProperties.getGenericErrorHandler());
}
else {
properties.setErrorHandler((ErrorHandler) this.containerProperties.getGenericErrorHandler());
if (this.errorHandler != null) {
instance.setGenericErrorHandler(this.errorHandler);
}
}

4
spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2016 the original author or authors.
* Copyright 2014-2018 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.
@@ -19,7 +19,7 @@ package org.springframework.kafka.config;
import java.util.Collection;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
/**

96
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
View File

@@ -32,7 +32,6 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -56,57 +55,6 @@ public abstract class AbstractMessageListenerContainer<K, V>
protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
/**
* The offset commit behavior enumeration.
*/
public enum AckMode {
/**
* Commit after each record is processed by the listener.
*/
RECORD,
/**
* Commit whatever has already been processed before the next poll.
*/
BATCH,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
*/
TIME,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded.
*/
COUNT,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded or after {@link ContainerProperties#setAckTime(long)
* ackTime} has elapsed.
*/
COUNT_TIME,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}.
*/
MANUAL,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}. The consumer
* immediately processes the commit.
*/
MANUAL_IMMEDIATE,
}
protected final ConsumerFactory<K, V> consumerFactory; // NOSONAR (final)
private final ContainerProperties containerProperties;
@@ -117,6 +65,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
private ApplicationEventPublisher applicationEventPublisher;
private GenericErrorHandler<?> errorHandler;
private boolean autoStartup = true;
private int phase = DEFAULT_PHASE;
@@ -168,12 +118,6 @@ public abstract class AbstractMessageListenerContainer<K, V>
if (this.containerProperties.getConsumerRebalanceListener() == null) {
this.containerProperties.setConsumerRebalanceListener(createSimpleLoggingConsumerRebalanceListener());
}
if (containerProperties.getGenericErrorHandler() instanceof BatchErrorHandler) {
this.containerProperties.setBatchErrorHandler((BatchErrorHandler) containerProperties.getGenericErrorHandler());
}
else {
this.containerProperties.setErrorHandler((ErrorHandler) containerProperties.getGenericErrorHandler());
}
}
@Override
@@ -194,6 +138,42 @@ public abstract class AbstractMessageListenerContainer<K, V>
return this.applicationEventPublisher;
}
/**
* Set the error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
* @since 2.2
*/
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Set the error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
* @since 2.2
*/
public void setGenericErrorHandler(GenericErrorHandler<?> errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Set the batch error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
* @since 2.2
*/
public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Get the configured error handler.
* @return the error handler.
* @since 2.2
*/
protected GenericErrorHandler<?> getGenericErrorHandler() {
return this.errorHandler;
}
@Override
public boolean isAutoStartup() {
return this.autoStartup;

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

@@ -32,7 +32,6 @@ import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.util.Assert;
@@ -161,6 +160,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
container.setApplicationEventPublisher(getApplicationEventPublisher());
}
container.setClientIdSuffix("-" + i);
container.setGenericErrorHandler(getGenericErrorHandler());
container.start();
this.containers.add(container);
}

90
spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java → spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.kafka.listener.config;
package org.springframework.kafka.listener;
import java.util.Arrays;
import java.util.LinkedHashSet;
@@ -24,11 +24,6 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.BatchErrorHandler;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.GenericErrorHandler;
import org.springframework.kafka.support.LogIfLevelEnabled;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.scheduling.TaskScheduler;
@@ -46,6 +41,57 @@ import org.springframework.util.StringUtils;
*/
public class ContainerProperties {
/**
* The offset commit behavior enumeration.
*/
public enum AckMode {
/**
* Commit after each record is processed by the listener.
*/
RECORD,
/**
* Commit whatever has already been processed before the next poll.
*/
BATCH,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
*/
TIME,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded.
*/
COUNT,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded or after {@link ContainerProperties#setAckTime(long)
* ackTime} has elapsed.
*/
COUNT_TIME,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}.
*/
MANUAL,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}. The consumer
* immediately processes the commit.
*/
MANUAL_IMMEDIATE,
}
private static final long DEFAULT_POLL_TIMEOUT = 1000L;
private static final int DEFAULT_SHUTDOWN_TIMEOUT = 10000;
@@ -82,7 +128,7 @@ public class ContainerProperties {
* {@link org.springframework.kafka.listener.AcknowledgingMessageListener}.
* </ul>
*/
private AbstractMessageListenerContainer.AckMode ackMode = AckMode.BATCH;
private AckMode ackMode = AckMode.BATCH;
/**
* The number of outstanding record count after which offsets should be
@@ -114,11 +160,6 @@ public class ContainerProperties {
*/
private AsyncListenableTaskExecutor consumerTaskExecutor;
/**
* The error handler to call when the listener throws an exception.
*/
private GenericErrorHandler<?> errorHandler;
/**
* The timeout for shutting down the container. This is the maximum amount of
* time that the invocation to {@code #stop(Runnable)} will block for, before
@@ -209,7 +250,7 @@ public class ContainerProperties {
* </ul>
* @param ackMode the {@link AckMode}; default BATCH.
*/
public void setAckMode(AbstractMessageListenerContainer.AckMode ackMode) {
public void setAckMode(AckMode ackMode) {
Assert.notNull(ackMode, "'ackMode' cannot be null");
this.ackMode = ackMode;
}
@@ -243,22 +284,6 @@ public class ContainerProperties {
this.ackTime = ackTime;
}
/**
* Set the error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
*/
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Set the batch error handler to call when the listener throws an exception.
* @param errorHandler the error handler.
*/
public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Set the executor for threads that poll the consumer.
* @param consumerTaskExecutor the executor
@@ -355,7 +380,7 @@ public class ContainerProperties {
return this.topicPartitions;
}
public AbstractMessageListenerContainer.AckMode getAckMode() {
public AckMode getAckMode() {
return this.ackMode;
}
@@ -379,10 +404,6 @@ public class ContainerProperties {
return this.consumerTaskExecutor;
}
public GenericErrorHandler<?> getGenericErrorHandler() {
return this.errorHandler;
}
public long getShutdownTimeout() {
return this.shutdownTimeout;
}
@@ -542,7 +563,6 @@ public class ContainerProperties {
+ ", pollTimeout=" + this.pollTimeout
+ (this.consumerTaskExecutor != null
? ", consumerTaskExecutor=" + this.consumerTaskExecutor : "")
+ (this.errorHandler != null ? ", errorHandler=" + this.errorHandler : "")
+ ", shutdownTimeout=" + this.shutdownTimeout
+ (this.consumerRebalanceListener != null
? ", consumerRebalanceListener=" + this.consumerRebalanceListener : "")

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

@@ -60,7 +60,7 @@ import org.springframework.kafka.event.ConsumerResumedEvent;
import org.springframework.kafka.event.ListenerContainerIdleEvent;
import org.springframework.kafka.event.NonResponsiveConsumerEvent;
import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.LogIfLevelEnabled;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
@@ -449,7 +449,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
consumer.assign(new ArrayList<>(this.definedPartitions.keySet()));
}
GenericErrorHandler<?> errHandler = this.containerProperties.getGenericErrorHandler();
GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
this.genericListener = listener;
if (listener instanceof BatchMessageListener) {
this.listener = null;
@@ -636,7 +636,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
private void validateErrorHandler(boolean batch) {
GenericErrorHandler<?> errHandler = this.containerProperties.getGenericErrorHandler();
GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
if (this.errorHandler == null) {
return;
}

1
spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java
View File

@@ -24,7 +24,6 @@ import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
import org.springframework.context.SmartLifecycle;
import org.springframework.kafka.listener.config.ContainerProperties;
/**
* Internal abstraction used by the framework representing a message

2
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java
View File

@@ -125,7 +125,7 @@ public class DelegatingInvocableHandler {
Object result = handler.invoke(message, providedArgs);
Expression replyTo = this.handlerSendTo.get(handler);
if (replyTo != null) {
result = new MessagingMessageListenerAdapter.ResultHolder(result, replyTo);
result = new InvocationResult(result, replyTo);
}
return result;
}

51
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/InvocationResult.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2018 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.adapter;
import org.springframework.expression.Expression;
/**
* The result of a method invocation.
*
* @author Gary Russell
* @since 2.2
*/
public final class InvocationResult {
private final Object result;
private final Expression sendTo;
public InvocationResult(Object result, Expression sendTo) {
this.result = result;
this.sendTo = sendTo;
}
public Object getResult() {
return this.result;
}
public Expression getSendTo() {
return this.sendTo;
}
@Override
public String toString() {
return this.result.toString();
}
}

28
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -281,7 +281,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
this.logger.debug("Listener method returned result [" + resultArg
+ "] - generating response message for it");
}
Object result = resultArg instanceof ResultHolder ? ((ResultHolder) resultArg).result : resultArg;
Object result = resultArg instanceof InvocationResult ? ((InvocationResult) resultArg).getResult() : resultArg;
String replyTopic = evaluateReplyTopic(request, source, resultArg);
Assert.state(replyTopic == null || this.replyTemplate != null,
"a KafkaTemplate is required to support replies");
@@ -290,8 +290,8 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
private String evaluateReplyTopic(Object request, Object source, Object result) {
String replyTo = null;
if (result instanceof ResultHolder) {
replyTo = evaluateTopic(request, source, result, ((ResultHolder) result).sendTo);
if (result instanceof InvocationResult) {
replyTo = evaluateTopic(request, source, result, ((InvocationResult) result).getSendTo());
}
else if (this.replyTopicExpression != null) {
replyTo = evaluateTopic(request, source, result, this.replyTopicExpression);
@@ -505,28 +505,6 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
return !parameterType.equals(Message.class); // could be Message without a generic type
}
/**
* Result holder.
* @since 2.0
*/
public static final class ResultHolder {
private final Object result;
private final Expression sendTo;
public ResultHolder(Object result, Expression sendTo) {
this.result = result;
this.sendTo = sendTo;
}
@Override
public String toString() {
return this.result.toString();
}
}
/**
* Root object for reply expression evaluation.
* @since 2.0

4
spring-kafka/src/main/java/org/springframework/kafka/listener/config/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Container configuration.
*/
package org.springframework.kafka.listener.config;

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

@@ -62,12 +62,13 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.event.ListenerContainerIdleEvent;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareErrorHandler;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.listener.MessageListenerContainer;
@@ -75,7 +76,6 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte
import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.KafkaNull;
@@ -669,7 +669,7 @@ public class EnableKafkaIntegrationTests {
factory.setConsumerFactory(consumerFactory());
factory.setRecordFilterStrategy(recordFilter());
factory.setReplyTemplate(partitionZeroReplyingTemplate());
factory.getContainerProperties().setErrorHandler((ConsumerAwareErrorHandler) (t, d, c) -> {
factory.setErrorHandler((ConsumerAwareErrorHandler) (t, d, c) -> {
this.globalErrorThrowable = t;
c.seek(new org.apache.kafka.common.TopicPartition(d.topic(), d.partition()), d.offset());
});
@@ -832,14 +832,15 @@ public class EnableKafkaIntegrationTests {
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
recordAckListenerContainerFactory() {
recordAckListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(manualConsumerFactory("clientIdViaProps4"));
ContainerProperties props = factory.getContainerProperties();
props.setAckMode(AckMode.RECORD);
props.setAckOnError(true);
props.setErrorHandler(listen16ErrorHandler());
factory.setErrorHandler(listen16ErrorHandler());
return factory;
}

2
spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java
View File

@@ -90,7 +90,7 @@ public class StatefulRetryTests {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler() {
factory.setErrorHandler(new SeekToCurrentErrorHandler() {
@Override
public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,

22
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -52,8 +52,6 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.ContainerTestUtils;
@@ -242,14 +240,14 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testManualCommit() throws Exception {
testManualCommitGuts(AckMode.MANUAL, topic4);
testManualCommitGuts(AckMode.MANUAL_IMMEDIATE, topic5);
testManualCommitGuts(ContainerProperties.AckMode.MANUAL, topic4);
testManualCommitGuts(ContainerProperties.AckMode.MANUAL_IMMEDIATE, topic5);
// to be sure the commits worked ok so run the tests again and the second tests start at the committed offset.
testManualCommitGuts(AckMode.MANUAL, topic4);
testManualCommitGuts(AckMode.MANUAL_IMMEDIATE, topic5);
testManualCommitGuts(ContainerProperties.AckMode.MANUAL, topic4);
testManualCommitGuts(ContainerProperties.AckMode.MANUAL_IMMEDIATE, topic5);
}
private void testManualCommitGuts(AckMode ackMode, String topic) throws Exception {
private void testManualCommitGuts(ContainerProperties.AckMode ackMode, String topic) throws Exception {
this.logger.info("Start " + ackMode);
Map<String, Object> props = KafkaTestUtils.consumerProps("test" + ackMode, "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -306,7 +304,7 @@ public class ConcurrentMessageListenerContainerTests {
ack.acknowledge();
latch.countDown();
});
containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE);
containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
containerProps.setSyncCommits(false);
final CountDownLatch commits = new CountDownLatch(8);
final AtomicReference<Exception> exceptionRef = new AtomicReference<>();
@@ -361,7 +359,7 @@ public class ConcurrentMessageListenerContainerTests {
bitSet.set((int) (message.partition() * 4 + message.offset()));
latch.countDown();
});
containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE);
containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
ConcurrentMessageListenerContainer<Integer, String> container =
new ConcurrentMessageListenerContainer<>(cf, containerProps);
@@ -440,12 +438,12 @@ public class ConcurrentMessageListenerContainerTests {
latch.countDown();
throw new RuntimeException("intended");
});
containerProps.setErrorHandler((thrownException, record) -> catchError.set(true));
ConcurrentMessageListenerContainer<Integer, String> container =
new ConcurrentMessageListenerContainer<>(cf, containerProps);
container.setConcurrency(2);
container.setBeanName("testException");
container.setErrorHandler((thrownException, record) -> catchError.set(true));
container.start();
ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
@@ -481,7 +479,7 @@ public class ConcurrentMessageListenerContainerTests {
}
});
containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.RECORD);
containerProps.setAckMode(ContainerProperties.AckMode.RECORD);
containerProps.setAckOnError(false);
ConcurrentMessageListenerContainer<Integer, String> container = new ConcurrentMessageListenerContainer<>(cf,
containerProps);
@@ -546,7 +544,7 @@ public class ConcurrentMessageListenerContainerTests {
final CountDownLatch latch = new CountDownLatch(2);
ContainerProperties containerProps = new ContainerProperties(topic);
containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE);
containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
containerProps.setAckOnError(ackOnError);
containerProps.setMessageListener((AcknowledgingMessageListener<Integer, String>) (message, ack) -> {
ConcurrentMessageListenerContainerTests.this.logger.info("manualExisting: " + message);

4
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -181,7 +181,7 @@ public class ContainerStoppingBatchErrorHandlerTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setBatchErrorHandler(new ContainerStoppingBatchErrorHandler() {
factory.setBatchErrorHandler(new ContainerStoppingBatchErrorHandler() {
@Override
public void handle(Exception thrownException, ConsumerRecords<?, ?> records,

7
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -52,7 +52,6 @@ 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;
@@ -187,7 +186,7 @@ public class ContainerStoppingErrorHandlerBatchModeTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(new ContainerStoppingErrorHandler() {
factory.setErrorHandler(new ContainerStoppingErrorHandler() {
@Override
public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
@@ -204,7 +203,7 @@ public class ContainerStoppingErrorHandlerBatchModeTests {
}
});
factory.getContainerProperties().setAckMode(AckMode.BATCH);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
return factory;
}

7
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -53,7 +53,6 @@ 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;
@@ -201,7 +200,7 @@ public class ContainerStoppingErrorHandlerRecordModeTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(new ContainerStoppingErrorHandler() {
factory.setErrorHandler(new ContainerStoppingErrorHandler() {
@Override
public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
@@ -218,7 +217,7 @@ public class ContainerStoppingErrorHandlerRecordModeTests {
}
});
factory.getContainerProperties().setAckMode(AckMode.RECORD);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
return factory;
}

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

@@ -76,9 +76,8 @@ import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.event.ConsumerPausedEvent;
import org.springframework.kafka.event.ConsumerResumedEvent;
import org.springframework.kafka.event.NonResponsiveConsumerEvent;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
@@ -925,17 +924,17 @@ public class KafkaMessageListenerContainerTests {
containerProps.setPollTimeout(100);
containerProps.setAckOnError(true);
final CountDownLatch latch = new CountDownLatch(4);
containerProps.setBatchErrorHandler((t, messages) -> {
new BatchLoggingErrorHandler().handle(t, messages);
for (int i = 0; i < messages.count(); i++) {
latch.countDown();
}
});
CountDownLatch stubbingComplete = new CountDownLatch(1);
KafkaMessageListenerContainer<Integer, String> container = spyOnContainer(
new KafkaMessageListenerContainer<>(cf, containerProps), stubbingComplete);
container.setBeanName("testBatchListenerErrors");
container.setBatchErrorHandler((t, messages) -> {
new BatchLoggingErrorHandler().handle(t, messages);
for (int i = 0; i < messages.count(); i++) {
latch.countDown();
}
});
container.start();
Consumer<?, ?> containerConsumer = spyOnConsumer(container);
final CountDownLatch commitLatch = new CountDownLatch(2);
@@ -1863,7 +1862,6 @@ public class KafkaMessageListenerContainerTests {
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);
@@ -1886,6 +1884,7 @@ public class KafkaMessageListenerContainerTests {
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
container.setBeanName("testContainerException");
container.setErrorHandler(new SeekToCurrentErrorHandler());
container.start();
ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
container.pause();

1
spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java
View File

@@ -34,7 +34,6 @@ import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.test.rule.KafkaEmbedded;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -194,7 +194,7 @@ public class SeekToCurrentBatchErrorHandlerTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
factory.setBatchListener(true);
factory.getContainerProperties().setTransactionManager(tm());
return factory;

6
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -55,7 +55,7 @@ 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.listener.ContainerProperties.AckMode;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -224,7 +224,7 @@ public class SeekToCurrentOnErrorBatchModeTXTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
factory.setErrorHandler(new SeekToCurrentErrorHandler());
factory.getContainerProperties().setAckMode(AckMode.BATCH);
factory.getContainerProperties().setTransactionManager(tm());
return factory;

6
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -53,7 +53,7 @@ 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.kafka.listener.ContainerProperties.AckMode;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -204,7 +204,7 @@ public class SeekToCurrentOnErrorBatchModeTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
factory.setErrorHandler(new SeekToCurrentErrorHandler());
factory.getContainerProperties().setAckMode(AckMode.BATCH);
return factory;
}

6
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -55,7 +55,7 @@ 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.listener.ContainerProperties.AckMode;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -230,7 +230,7 @@ public class SeekToCurrentOnErrorRecordModeTXTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
factory.setErrorHandler(new SeekToCurrentErrorHandler());
factory.getContainerProperties().setAckMode(AckMode.RECORD);
factory.getContainerProperties().setTransactionManager(tm());
return factory;

6
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -53,7 +53,7 @@ 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.kafka.listener.ContainerProperties.AckMode;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -208,7 +208,7 @@ public class SeekToCurrentOnErrorRecordModeTests {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
factory.setErrorHandler(new SeekToCurrentErrorHandler());
factory.getContainerProperties().setAckMode(AckMode.RECORD);
return factory;
}

1
spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java
View File

@@ -61,7 +61,6 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.kafka.transaction.ChainedKafkaTransactionManager;

2
spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java
View File

@@ -47,8 +47,8 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SimpleKafkaHeaderMapper;
import org.springframework.kafka.support.converter.MessagingMessageConverter;

60
src/reference/asciidoc/changes-since-1.0.adoc
View File

@@ -1,4 +1,64 @@
[[migration]]
=== Changes between 2.0 and 2.1
==== Kafka Client Version
This version requires the 1.0.0 `kafka-clients` or higher.
NOTE: The 1.1.x client is supported, with _version 2.1.5_, but you will need to override dependencies as described in <<deps-for-11x>>.
The 1.1.x client will be supported natively in _version 2.2_.
==== JSON Improvements
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.
==== Pausing/Resuming Containers
The listener containers now have `pause()` and `resume()` methods (since _version 2.1.3_).
See <<pause-resume>> for more information.
==== Stateful Retry
Starting with _version 2.1.3_, stateful retry can be configured; see <<stateful-retry>> for more information.
==== Client ID
Starting with _version 2.1.1_, it is now possible to set the `client.id` prefix on `@KafkaListener`.
Previously, to customize the client id, you would need a separate consumer factory (and container factory) per listener.
The prefix is suffixed with `-n` to provide unique client ids when using concurrency.
==== Logging Offset Commits
By default, logging of topic offset commits is performed with the DEBUG logging level.
Starting with _version 2.1.2_, there is a new property in `ContainerProperties` called `commitLogLevel` which allows you to specify the log level for these messages.
See <<kafka-container>> for more information.
==== Default @KafkaHandler
Starting with _version 2.1.3_, one of the `@KafkaHandler` s on a class-level `@KafkaListener` can be designated as the default.
See <<class-level-kafkalistener>> for more information.
==== ReplyingKafkaTemplate
Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics.
See <<replying-template>> for more information.
==== ChainedKafkaTransactionManager
_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <<chained-transaction-manager>> for more information.
==== Migration Guide from 2.0
https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].
=== Changes Between 1.3 and 2.0
==== Spring Framework and Java Versions

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

@@ -2,58 +2,12 @@
==== Kafka Client Version
This version requires the 1.0.0 `kafka-clients` or higher.
This version requires the 1.1.0 `kafka-clients` or higher.
NOTE: The 1.1.x client is supported, with _version 2.1.5_, but you will need to override dependencies as described in <<deps-for-11x>>.
The 1.1.x client will be supported natively in _version 2.2_.
==== Class/Package Changes
==== JSON Improvements
The class `ContainerProperties` has been moved from `org.springframework.kafka.listener.config` to `org.springframework.kafka.listener`.
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.
The enum `AckMode` has been moved from `AbstractMessageListenerContainer` to `ContainerProperties`.
==== 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.
==== Pausing/Resuming Containers
The listener containers now have `pause()` and `resume()` methods (since _version 2.1.3_).
See <<pause-resume>> for more information.
==== Stateful Retry
Starting with _version 2.1.3_, stateful retry can be configured; see <<stateful-retry>> for more information.
==== Client ID
Starting with _version 2.1.1_, it is now possible to set the `client.id` prefix on `@KafkaListener`.
Previously, to customize the client id, you would need a separate consumer factory (and container factory) per listener.
The prefix is suffixed with `-n` to provide unique client ids when using concurrency.
==== Logging Offset Commits
By default, logging of topic offset commits is performed with the DEBUG logging level.
Starting with _version 2.1.2_, there is a new property in `ContainerProperties` called `commitLogLevel` which allows you to specify the log level for these messages.
See <<kafka-container>> for more information.
==== Default @KafkaHandler
Starting with _version 2.1.3_, one of the `@KafkaHandler` s on a class-level `@KafkaListener` can be designated as the default.
See <<class-level-kafkalistener>> for more information.
==== ReplyingKafkaTemplate
Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics.
See <<replying-template>> for more information.
==== ChainedKafkaTransactionManager
_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <<chained-transaction-manager>> for more information.
==== Migration Guide from 2.0
https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].
`setBatchErrorHandler()` and `setErrorHandler()` methods have been moved from `ContainterProperties` to `AbstractMessageListenerContainer` (and `AbstractKafkaListenerContainerFactory`).