From b048aaa8f00045e4cc65d5d137b1aa372beca3a2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 3 Apr 2018 16:07:26 -0400 Subject: [PATCH] GH-550: master to 2.2; fix tangles 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 --- gradle.properties | 2 +- ...AbstractKafkaListenerContainerFactory.java | 30 ++++-- ...ncurrentKafkaListenerContainerFactory.java | 4 +- .../AbstractMessageListenerContainer.java | 96 ++++++++----------- .../ConcurrentMessageListenerContainer.java | 2 +- .../{config => }/ContainerProperties.java | 90 ++++++++++------- .../KafkaMessageListenerContainer.java | 6 +- .../listener/MessageListenerContainer.java | 1 - .../adapter/DelegatingInvocableHandler.java | 2 +- .../listener/adapter/InvocationResult.java | 51 ++++++++++ .../MessagingMessageListenerAdapter.java | 28 +----- .../kafka/listener/config/package-info.java | 4 - .../EnableKafkaIntegrationTests.java | 11 ++- .../kafka/annotation/StatefulRetryTests.java | 2 +- ...ncurrentMessageListenerContainerTests.java | 22 ++--- ...ntainerStoppingBatchErrorHandlerTests.java | 4 +- ...nerStoppingErrorHandlerBatchModeTests.java | 7 +- ...erStoppingErrorHandlerRecordModeTests.java | 7 +- .../KafkaMessageListenerContainerTests.java | 17 ++-- .../kafka/listener/MissingGroupIdTests.java | 1 - .../SeekToCurrentBatchErrorHandlerTests.java | 4 +- .../SeekToCurrentOnErrorBatchModeTXTests.java | 6 +- .../SeekToCurrentOnErrorBatchModeTests.java | 6 +- ...SeekToCurrentOnErrorRecordModeTXTests.java | 6 +- .../SeekToCurrentOnErrorRecordModeTests.java | 6 +- .../listener/TransactionalContainerTests.java | 1 - .../ReplyingKafkaTemplateTests.java | 2 +- src/reference/asciidoc/changes-since-1.0.adoc | 60 ++++++++++++ src/reference/asciidoc/whats-new.adoc | 56 +---------- 29 files changed, 292 insertions(+), 242 deletions(-) rename spring-kafka/src/main/java/org/springframework/kafka/listener/{config => }/ContainerProperties.java (92%) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/InvocationResult.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/config/package-info.java diff --git a/gradle.properties b/gradle.properties index bbf2bb4a..f212918a 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1 +1 @@ -version=2.1.6.BUILD-SNAPSHOT +version=2.2.0.BUILD-SNAPSHOT diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index f083ed10..fae6091e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -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 errorHandler; + private ConsumerFactory consumerFactory; private Boolean autoStartup; @@ -191,6 +194,24 @@ public abstract class AbstractKafkaListenerContainerFactory 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); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java index 6404c268..11411892 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java @@ -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; /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index da81b5d9..4b684c30 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -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 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 consumerFactory; // NOSONAR (final) private final ContainerProperties containerProperties; @@ -117,6 +65,8 @@ public abstract class AbstractMessageListenerContainer private ApplicationEventPublisher applicationEventPublisher; + private GenericErrorHandler errorHandler; + private boolean autoStartup = true; private int phase = DEFAULT_PHASE; @@ -168,12 +118,6 @@ public abstract class AbstractMessageListenerContainer 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 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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index e38d6a7c..d414e484 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -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 extends AbstractMessageLis container.setApplicationEventPublisher(getApplicationEventPublisher()); } container.setClientIdSuffix("-" + i); + container.setGenericErrorHandler(getGenericErrorHandler()); container.start(); this.containers.add(container); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java similarity index 92% rename from spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java rename to spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java index 1f02a691..4e7c1768 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java @@ -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}. * */ - 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 { * * @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 : "") 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 a18051cb..73262017 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 @@ -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 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 extends AbstractMessageListener } private void validateErrorHandler(boolean batch) { - GenericErrorHandler errHandler = this.containerProperties.getGenericErrorHandler(); + GenericErrorHandler errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler(); if (this.errorHandler == null) { return; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java index c9cc096f..fd39ee79 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java @@ -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 diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java index 18ea4bec..c240ecb4 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java @@ -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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/InvocationResult.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/InvocationResult.java new file mode 100644 index 00000000..a8e81bcd --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/InvocationResult.java @@ -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(); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index c84a1c63..baa07280 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java @@ -281,7 +281,7 @@ public abstract class MessagingMessageListenerAdapter 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 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 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 diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/package-info.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/package-info.java deleted file mode 100644 index f5735979..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -/** - * Container configuration. - */ -package org.springframework.kafka.listener.config; 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 9c1ba2f5..90c205d3 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 @@ -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> - recordAckListenerContainerFactory() { + recordAckListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory 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; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java index 452cecbc..cd77e72b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java @@ -90,7 +90,7 @@ public class StatefulRetryTests { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); - factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler() { + factory.setErrorHandler(new SeekToCurrentErrorHandler() { @Override public void handle(Exception thrownException, List> records, diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 9728f083..cb24fca8 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -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 props = KafkaTestUtils.consumerProps("test" + ackMode, "false", embeddedKafka); DefaultKafkaConsumerFactory 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 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 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 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 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) (message, ack) -> { ConcurrentMessageListenerContainerTests.this.logger.info("manualExisting: " + message); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java index c1dd6290..42ae9544 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java @@ -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, diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java index bcd8ad8a..0caf193f 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java @@ -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> records, @@ -204,7 +203,7 @@ public class ContainerStoppingErrorHandlerBatchModeTests { } }); - factory.getContainerProperties().setAckMode(AckMode.BATCH); + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH); return factory; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java index 96a5729c..b1063509 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java @@ -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> records, @@ -218,7 +217,7 @@ public class ContainerStoppingErrorHandlerRecordModeTests { } }); - factory.getContainerProperties().setAckMode(AckMode.RECORD); + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD); return factory; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index beb33be2..8128d654 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -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 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 senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); @@ -1886,6 +1884,7 @@ public class KafkaMessageListenerContainerTests { KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); container.setBeanName("testContainerException"); + container.setErrorHandler(new SeekToCurrentErrorHandler()); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); container.pause(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java index 9b6fbd67..5c2ca0f1 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java @@ -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; 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 9a153c28..77ca4b26 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 @@ -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; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java index 9f2ce160..d809564c 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java @@ -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; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java index 6bd89a26..f9c54677 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java @@ -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; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java index d897005e..ee2cf2fd 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java @@ -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; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java index b7dd8141..0a5af751 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java @@ -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; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index 69184e9f..661f759f 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java @@ -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; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java index d9d49338..716d3467 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java @@ -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; diff --git a/src/reference/asciidoc/changes-since-1.0.adoc b/src/reference/asciidoc/changes-since-1.0.adoc index ce35fdfb..9f934aeb 100644 --- a/src/reference/asciidoc/changes-since-1.0.adoc +++ b/src/reference/asciidoc/changes-since-1.0.adoc @@ -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 <>. +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 <> 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 <> for more information. + +==== Pausing/Resuming Containers + +The listener containers now have `pause()` and `resume()` methods (since _version 2.1.3_). +See <> for more information. + +==== Stateful Retry + +Starting with _version 2.1.3_, stateful retry can be configured; see <> 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 <> 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 <> for more information. + +==== ReplyingKafkaTemplate + +Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics. +See <> for more information. + +==== ChainedKafkaTransactionManager + +_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <> 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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index deab10d6..163b8a40 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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 <>. -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 <> 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 <> for more information. - -==== Pausing/Resuming Containers - -The listener containers now have `pause()` and `resume()` methods (since _version 2.1.3_). -See <> for more information. - -==== Stateful Retry - -Starting with _version 2.1.3_, stateful retry can be configured; see <> 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 <> 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 <> for more information. - -==== ReplyingKafkaTemplate - -Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics. -See <> for more information. - -==== ChainedKafkaTransactionManager - -_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <> 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`).