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

Remove deprecated interfaces, classes, and methods

Browse Source
This commit is contained in:
Gary Russell
2023-09-06 10:26:28 -07:00
committed by GitHub
parent 765efdcbba
commit ea643a90f4
51 changed files with 196 additions and 2035 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
spring-kafka-docs/src/main/asciidoc/changes-since-1.0.adoc
View File

@@ -1,4 +1,113 @@
[[migration]]
=== What's New in 3.0 Since 2.9
[[x30-kafka-client]]
==== Kafka Client Version
This version requires the 3.3.1 `kafka-clients`.
[[x30-eos]]
==== Exactly Once Semantics
`EOSMode.V1` (aka `ALPHA`) is no longer supported.
IMPORTANT: When using transactions, the minimum broker version is 2.5.
See <<exactly-once>> and https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
[[x30-obs]]
==== Observation
Enabling observation for timers and tracing using Micrometer is now supported.
See <<observation>> for more information.
[[x30-Native]]
==== Native Images
Support for creating native images is provided.
See <<native-images>> for more information.
[[x30-global-embedded-kafka]]
==== Global Single Embedded Kafka
The embedded Kafka (`EmbeddedKafkaBroker`) can now be start as a single global instance for the whole test plan.
See <<same-broker-multiple-tests>> for more information.
[[x30-retryable]]
==== Retryable Topics Changes
This feature is no longer considered experimental (as far as its API is concerned), the feature itself has been supported since 2.7, but with a greater than normal possibility of breaking API changes.
The bootstrapping of <<retry-topic>> infrastructure beans has changed in this release to avoid some timing problems that occurred in some application regarding application initialization.
You can now set a different `concurrency` for the retry containers; by default, the concurrency is the same as the main container.
`@RetryableTopic` can now be used as a meta-annotation on custom annotations, including support for `@AliasFor` properties.
See <<retry-config>> for more information.
The default replication factor for the retry topics is now `-1` (use broker default).
If your broker is earlier that version 2.4, you will now need to explicitly set the property.
You can now configure multiple `@RetryableTopic` listeners on the same topic in the same application context.
Previously, this was not possible.
See <<multi-retry>> for more information.
There are breaking API changes in `RetryTopicConfigurationSupport`; specifically, if you override the bean definition methods for `destinationTopicResolver`, `kafkaConsumerBackoffManager` and/or `retryTopicConfigurer`;
these methods now require an `ObjectProvider<RetryTopicComponentFactory>` parameter.
[[x30-lc-changes]]
==== Listener Container Changes
Events related to consumer authentication and authorization failures are now published by the container.
See <<events>> for more information.
You can now customize the thread names used by consumer threads.
See <<container-thread-naming>> for more information.
The container property `restartAfterAuthException` has been added.
See <<container-props>> for more information.
[[x30-template-changes]]
==== `KafkaTemplate` Changes
The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s.
See <<kafka-template>>.
[[x30-rkt-changes]]
==== `ReplyingKafkaTemplate` Changes
The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s.
See <<replying-template>> and <<exchanging-messages>>.
[[x30-listener]]
==== `@KafkaListener` Changes
You can now use a custom correlation header which will be echoed in any reply message.
See the note at the end of <<replying-template>> for more information.
You can now manually commit parts of a batch before the entire batch is processed.
See <<committing-offsets>> for more information.
[[x30-headers]]
==== `KafkaHeaders` Changes
Four constants in `KafkaHeaders` that were deprecated in 2.9.x have now been removed.
* Instead of `MESSAGE_KEY`, use `KEY`.
* Instead of `PARTITION_ID`, use `PARTITION`
Similarly, `RECEIVED_MESSAGE_KEY` is replaced by `RECEIVED_KEY` and `RECEIVED_PARTITION_ID` is replaced by `RECEIVED_PARTITION`.
[[x30-testing]]
==== Testing Changes
Version 3.0.7 introduced a `MockConsumerFactory` and `MockProducerFactory`.
See <<mock-cons-prod>> for more information.
Starting with version 3.0.10, the embedded Kafka broker, by default, sets the Spring Boot property `spring.kafka.bootstrap-servers` to the address(es) of the embedded broker(s).
=== What's New in 2.9 since 2.8
[[x29-kafka-client]]

107
spring-kafka-docs/src/main/asciidoc/whats-new.adoc
View File

@@ -1,111 +1,10 @@
=== What's New in 3.0 Since 2.9
=== What's New in 3.1 Since 3.0
This section covers the changes made from version 2.9 to version 3.0.
This section covers the changes made from version 3.0 to version 3.1.
For changes in earlier version, see <<history>>.
[[x30-kafka-client]]
==== Kafka Client Version
This version requires the 3.3.1 `kafka-clients`.
This version requires the 3.5.1 `kafka-clients`.
[[x30-eos]]
==== Exactly Once Semantics
`EOSMode.V1` (aka `ALPHA`) is no longer supported.
IMPORTANT: When using transactions, the minimum broker version is 2.5.
See <<exactly-once>> and https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
[[x30-obs]]
==== Observation
Enabling observation for timers and tracing using Micrometer is now supported.
See <<observation>> for more information.
[[x30-Native]]
==== Native Images
Support for creating native images is provided.
See <<native-images>> for more information.
[[x30-global-embedded-kafka]]
==== Global Single Embedded Kafka
The embedded Kafka (`EmbeddedKafkaBroker`) can now be start as a single global instance for the whole test plan.
See <<same-broker-multiple-tests>> for more information.
[[x30-retryable]]
==== Retryable Topics Changes
This feature is no longer considered experimental (as far as its API is concerned), the feature itself has been supported since 2.7, but with a greater than normal possibility of breaking API changes.
The bootstrapping of <<retry-topic>> infrastructure beans has changed in this release to avoid some timing problems that occurred in some application regarding application initialization.
You can now set a different `concurrency` for the retry containers; by default, the concurrency is the same as the main container.
`@RetryableTopic` can now be used as a meta-annotation on custom annotations, including support for `@AliasFor` properties.
See <<retry-config>> for more information.
The default replication factor for the retry topics is now `-1` (use broker default).
If your broker is earlier that version 2.4, you will now need to explicitly set the property.
You can now configure multiple `@RetryableTopic` listeners on the same topic in the same application context.
Previously, this was not possible.
See <<multi-retry>> for more information.
There are breaking API changes in `RetryTopicConfigurationSupport`; specifically, if you override the bean definition methods for `destinationTopicResolver`, `kafkaConsumerBackoffManager` and/or `retryTopicConfigurer`;
these methods now require an `ObjectProvider<RetryTopicComponentFactory>` parameter.
[[x30-lc-changes]]
==== Listener Container Changes
Events related to consumer authentication and authorization failures are now published by the container.
See <<events>> for more information.
You can now customize the thread names used by consumer threads.
See <<container-thread-naming>> for more information.
The container property `restartAfterAuthException` has been added.
See <<container-props>> for more information.
[[x30-template-changes]]
==== `KafkaTemplate` Changes
The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s.
See <<kafka-template>>.
[[x30-rkt-changes]]
==== `ReplyingKafkaTemplate` Changes
The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s.
See <<replying-template>> and <<exchanging-messages>>.
[[x30-listener]]
==== `@KafkaListener` Changes
You can now use a custom correlation header which will be echoed in any reply message.
See the note at the end of <<replying-template>> for more information.
You can now manually commit parts of a batch before the entire batch is processed.
See <<committing-offsets>> for more information.
[[x30-headers]]
==== `KafkaHeaders` Changes
Four constants in `KafkaHeaders` that were deprecated in 2.9.x have now been removed.
* Instead of `MESSAGE_KEY`, use `KEY`.
* Instead of `PARTITION_ID`, use `PARTITION`
Similarly, `RECEIVED_MESSAGE_KEY` is replaced by `RECEIVED_KEY` and `RECEIVED_PARTITION_ID` is replaced by `RECEIVED_PARTITION`.
[[x30-testing]]
==== Testing Changes
Version 3.0.7 introduced a `MockConsumerFactory` and `MockProducerFactory`.
See <<mock-cons-prod>> for more information.
Starting with version 3.0.10, the embedded Kafka broker, by default, sets the Spring Boot property `spring.kafka.bootstrap-servers` to the address(es) of the embedded broker(s).

8
spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java
View File

@@ -196,14 +196,6 @@ public @interface RetryableTopic {
*/
DltStrategy dltStrategy() default DltStrategy.ALWAYS_RETRY_ON_ERROR;
/**
* Whether to use a single or multiple topics when using a fixed delay.
* @return the fixed delay strategy.
* @deprecated in favor of {@link #sameIntervalTopicReuseStrategy()}.
*/
@Deprecated
org.springframework.kafka.retrytopic.FixedDelayStrategy fixedDelayTopicStrategy() default org.springframework.kafka.retrytopic.FixedDelayStrategy.MULTIPLE_TOPICS;
/**
* Override the container factory's {@code autoStartup} property for just the DLT container.
* Usually used to not start the DLT container when {@code autoStartup} is true.

1
spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java
View File

@@ -143,7 +143,6 @@ public class RetryableTopicAnnotationProcessor {
.retryOn(includes)
.notRetryOn(excludes)
.traversingCauses(traverse)
.useSingleTopicForFixedDelays(annotation.fixedDelayTopicStrategy())
.dltProcessingFailureStrategy(annotation.dltStrategy())
.autoStartDltHandler(autoStartDlt)
.setTopicSuffixingStrategy(annotation.topicSuffixingStrategy())

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

@@ -26,7 +26,6 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
@@ -48,7 +47,6 @@ import org.springframework.kafka.requestreply.ReplyingKafkaOperations;
import org.springframework.kafka.support.JavaUtils;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.util.Assert;
@@ -66,16 +64,12 @@ import org.springframework.util.Assert;
* @see AbstractMessageListenerContainer
*/
public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K, V>, K, V>
implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, InitializingBean,
ApplicationContextAware {
implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, ApplicationContextAware {
protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR protected
private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null); // NOSONAR
@SuppressWarnings("deprecation")
private org.springframework.kafka.listener.GenericErrorHandler<?> errorHandler;
private CommonErrorHandler commonErrorHandler;
private ConsumerFactory<? super K, ? super V> consumerFactory;
@@ -155,23 +149,6 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
this.phase = phase;
}
/**
* Set the message converter to use if dynamic argument type matching is needed.
* @param messageConverter the converter.
* @deprecated since 2.9.6 in favor of
* {@link #setBatchMessageConverter(BatchMessageConverter)} and
* {@link #setRecordMessageConverter(RecordMessageConverter)}.
*/
@Deprecated
public void setMessageConverter(MessageConverter messageConverter) {
if (messageConverter instanceof RecordMessageConverter) {
setRecordMessageConverter((RecordMessageConverter) messageConverter);
}
else {
setBatchMessageConverter((BatchMessageConverter) messageConverter);
}
}
/**
* Set the message converter to use if dynamic argument type matching is needed for
* record listeners.
@@ -247,30 +224,6 @@ 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
* @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
* @see #setCommonErrorHandler(CommonErrorHandler)
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
public void setErrorHandler(org.springframework.kafka.listener.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
* @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
* @see #setCommonErrorHandler(CommonErrorHandler)
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
public void setBatchErrorHandler(org.springframework.kafka.listener.BatchErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Set the {@link CommonErrorHandler} which can handle errors for both record and
* batch listeners. Replaces the use of
@@ -396,23 +349,6 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
this.threadNameSupplier = threadNameSupplier;
}
@SuppressWarnings("deprecation")
@Override
public void afterPropertiesSet() {
if (this.commonErrorHandler == null && this.errorHandler != null) {
if (Boolean.TRUE.equals(this.batchListener)) {
Assert.state(this.errorHandler instanceof org.springframework.kafka.listener.BatchErrorHandler,
() -> "The error handler must be a BatchErrorHandler, not " +
this.errorHandler.getClass().getName());
}
else {
Assert.state(this.errorHandler instanceof org.springframework.kafka.listener.ErrorHandler,
() -> "The error handler must be an ErrorHandler, not " +
this.errorHandler.getClass().getName());
}
}
}
@SuppressWarnings("unchecked")
@Override
public C createListenerContainer(KafkaListenerEndpoint endpoint) {
@@ -479,7 +415,6 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
properties::setAckTime)
.acceptIfNotNull(this.containerProperties.getSubBatchPerPartition(),
properties::setSubBatchPerPartition)
.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
.acceptIfNotNull(this.commonErrorHandler, instance::setCommonErrorHandler)
.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal)
.acceptIfNotNull(this.changeConsumerThreadName, instance::setChangeConsumerThreadName)

28
spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
View File

@@ -61,7 +61,6 @@ import org.springframework.context.SmartLifecycle;
import org.springframework.context.event.ContextStoppedEvent;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.listener.ContainerProperties.EOSMode;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -439,33 +438,6 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
return this.producerPerThread;
}
/**
* This is no longer needed now that only {@link EOSMode#V2} is supported. Ignored.
* @param producerPerConsumerPartition false to revert.
* @since 1.3.7
* @deprecated no longer necessary because
* {@code org.springframework.kafka.listener.ContainerProperties.EOSMode#V1} is no
* longer supported.
*/
@Deprecated(since = "3.0", forRemoval = true) // in 3.1
public void setProducerPerConsumerPartition(boolean producerPerConsumerPartition) {
}
/**
* This is no longer needed now that only {@link EOSMode#V2} is supported. Ignored.
* @return the producerPerConsumerPartition.
* @since 1.3.8
* @deprecated no longer necessary because
* {@code org.springframework.kafka.listener.ContainerProperties.EOSMode#V1} is no
* longer supported.
*/
@Deprecated(since = "3.0", forRemoval = true) // in 3.1
@Override
public boolean isProducerPerConsumerPartition() {
return false;
}
@Override
@Nullable
public Serializer<K> getKeySerializer() {

281
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations2.java
View File

@@ -1,281 +0,0 @@
/*
* Copyright 2015-2022 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
*
* https://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.core;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.core.KafkaOperations.OperationsCallback;
import org.springframework.kafka.core.KafkaOperations.ProducerCallback;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
/**
* The basic Kafka operations contract returning {@link CompletableFuture}s.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Gary Russell
* @since 2.9
* @deprecated no longer needed; use {@code KafkaOperations}.
*/
@Deprecated(since = "3.0", forRemoval = true) // in 3.1
public interface KafkaOperations2<K, V> {
/**
* Default timeout for {@link #receive(String, int, long)}.
*/
Duration DEFAULT_POLL_TIMEOUT = Duration.ofSeconds(5);
/**
* Send the data to the default topic with no key or partition.
* @param data The data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> sendDefault(V data);
/**
* Send the data to the default topic with the provided key and no partition.
* @param key the key.
* @param data The data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);
/**
* Send the data to the default topic with the provided key and partition.
* @param partition the partition.
* @param key the key.
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
/**
* Send the data to the default topic with the provided key and partition.
* @param partition the partition.
* @param timestamp the timestamp of the record.
* @param key the key.
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
/**
* Send the data to the provided topic with no key or partition.
* @param topic the topic.
* @param data The data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> send(String topic, V data);
/**
* Send the data to the provided topic with the provided key and no partition.
* @param topic the topic.
* @param key the key.
* @param data The data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);
/**
* Send the data to the provided topic with the provided key and partition.
* @param topic the topic.
* @param partition the partition.
* @param key the key.
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
/**
* Send the data to the provided topic with the provided key and partition.
* @param topic the topic.
* @param partition the partition.
* @param timestamp the timestamp of the record.
* @param key the key.
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
/**
* Send the provided {@link ProducerRecord}.
* @param record the record.
* @return a Future for the {@link SendResult}.
*/
CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
/**
* Send a message with routing information in message headers. The message payload
* may be converted before sending.
* @param message the message to send.
* @return a Future for the {@link SendResult}.
* @see org.springframework.kafka.support.KafkaHeaders#TOPIC
* @see org.springframework.kafka.support.KafkaHeaders#PARTITION
* @see org.springframework.kafka.support.KafkaHeaders#KEY
*/
CompletableFuture<SendResult<K, V>> send(Message<?> message);
/**
* See {@link Producer#partitionsFor(String)}.
* @param topic the topic.
* @return the partition info.
*/
List<PartitionInfo> partitionsFor(String topic);
/**
* See {@link Producer#metrics()}.
* @return the metrics.
*/
Map<MetricName, ? extends Metric> metrics();
/**
* Execute some arbitrary operation(s) on the producer and return the result.
* @param callback the callback.
* @param <T> the result type.
* @return the result.
*/
@Nullable
<T> T execute(ProducerCallback<K, V, T> callback);
/**
* Execute some arbitrary operation(s) on the operations and return the result.
* The operations are invoked within a local transaction and do not participate
* in a global transaction (if present).
* @param callback the callback.
* @param <T> the result type.
* @return the result.
*/
@Nullable
<T> T executeInTransaction(OperationsCallback<K, V, T> callback);
/**
* Flush the producer.
*/
void flush();
/**
* When running in a transaction, send the consumer offset(s) to the transaction. It
* is not necessary to call this method if the operations are invoked on a listener
* container thread (and the listener container is configured with a
* {@link org.springframework.kafka.transaction.KafkaAwareTransactionManager}) since
* the container will take care of sending the offsets to the transaction.
* Use with 2.5 brokers or later.
* @param offsets The offsets.
* @param groupMetadata the consumer group metadata.
* @see Producer#sendOffsetsToTransaction(Map, ConsumerGroupMetadata)
*/
default void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
ConsumerGroupMetadata groupMetadata) {
throw new UnsupportedOperationException();
}
/**
* Return true if the implementation supports transactions (has a transaction-capable
* producer factory).
* @return true or false.
*/
boolean isTransactional();
/**
* Return true if this template, when transactional, allows non-transactional operations.
* @return true to allow.
*/
default boolean isAllowNonTransactional() {
return false;
}
/**
* Return true if the template is currently running in a transaction on the calling
* thread.
* @return true if a transaction is running.
*/
default boolean inTransaction() {
return false;
}
/**
* Return the producer factory used by this template.
* @return the factory.
*/
default ProducerFactory<K, V> getProducerFactory() {
throw new UnsupportedOperationException("This implementation does not support this operation");
}
/**
* Receive a single record with the default poll timeout (5 seconds).
* @param topic the topic.
* @param partition the partition.
* @param offset the offset.
* @return the record or null.
* @see #DEFAULT_POLL_TIMEOUT
*/
@Nullable
default ConsumerRecord<K, V> receive(String topic, int partition, long offset) {
return receive(topic, partition, offset, DEFAULT_POLL_TIMEOUT);
}
/**
* Receive a single record.
* @param topic the topic.
* @param partition the partition.
* @param offset the offset.
* @param pollTimeout the timeout.
* @return the record or null.
*/
@Nullable
ConsumerRecord<K, V> receive(String topic, int partition, long offset, Duration pollTimeout);
/**
* Receive a multiple records with the default poll timeout (5 seconds). Only
* absolute, positive offsets are supported.
* @param requested a collection of record requests (topic/partition/offset).
* @return the records
* @see #DEFAULT_POLL_TIMEOUT
*/
default ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested) {
return receive(requested, DEFAULT_POLL_TIMEOUT);
}
/**
* Receive multiple records. Only absolute, positive offsets are supported.
* @param requested a collection of record requests (topic/partition/offset).
* @param pollTimeout the timeout.
* @return the record or null.
*/
ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested, Duration pollTimeout);
}

23
spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java
View File

@@ -78,29 +78,6 @@ public interface ProducerFactory<K, V> {
return false;
}
/**
* Remove the specified producer from the cache and close it.
* @param transactionIdSuffix the producer's transaction id suffix.
* @since 1.3.8
* @deprecated - no longer needed.
*/
@Deprecated(since = "3.0", forRemoval = true) // in 3.1
default void closeProducerFor(String transactionIdSuffix) {
}
/**
* Return the producerPerConsumerPartition.
* @return the producerPerConsumerPartition.
* @since 1.3.8
* @deprecated no longer necessary because
* {@code org.springframework.kafka.listener.ContainerProperties.EOSMode#V1} is no
* longer supported.
*/
@Deprecated(since = "3.0", forRemoval = true) // in 3.1
default boolean isProducerPerConsumerPartition() {
return false;
}
/**
* If the factory implementation uses thread-bound producers, call this method to
* close and release this thread's producer.

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

@@ -72,8 +72,6 @@ public abstract class AbstractMessageListenerContainer<K, V>
implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware,
ApplicationContextAware {
private static final String VERSION_2_8 = "2.8";
/**
* The default {@link org.springframework.context.SmartLifecycle} phase for listener
* containers {@value #DEFAULT_PHASE}.
@@ -98,9 +96,6 @@ public abstract class AbstractMessageListenerContainer<K, V>
private ApplicationEventPublisher applicationEventPublisher;
@SuppressWarnings("deprecation")
private GenericErrorHandler<?> errorHandler;
private CommonErrorHandler commonErrorHandler;
private boolean autoStartup = true;
@@ -227,55 +222,6 @@ 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
* @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
* @see #setCommonErrorHandler(CommonErrorHandler)
*/
@Deprecated(since = VERSION_2_8, forRemoval = true) // in 3.1
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
* @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
* @see #setCommonErrorHandler(CommonErrorHandler)
*/
@Deprecated(since = VERSION_2_8, forRemoval = true) // in 3.1
public void setGenericErrorHandler(@Nullable 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
* @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
* @see #setCommonErrorHandler(CommonErrorHandler)
*/
@Deprecated(since = VERSION_2_8, forRemoval = true) // in 3.1
public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Get the configured error handler.
* @return the error handler.
* @since 2.2
* @deprecated in favor of {@link #getCommonErrorHandler()}
* @see #getCommonErrorHandler()
*/
@Deprecated(since = VERSION_2_8, forRemoval = true) // in 3.1
@Nullable
public GenericErrorHandler<?> getGenericErrorHandler() {
return this.errorHandler;
}
/**
* Get the {@link CommonErrorHandler}.
* @return the handler.

64
spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java
View File

@@ -1,64 +0,0 @@
/*
* Copyright 2015-2022 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
*
* https://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;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.lang.Nullable;
/**
* Handles errors thrown during the execution of a {@link BatchMessageListener}.
* The listener should communicate which position(s) in the list failed in the
* exception.
*
* @author Gary Russell
*
* @since 1.1
* @deprecated in favor of {@link CommonErrorHandler}.
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
public interface BatchErrorHandler extends GenericErrorHandler<ConsumerRecords<?, ?>> {
/**
* Handle the exception.
* @param thrownException the exception.
* @param data the consumer records.
* @param consumer the consumer.
* @param container the container.
*/
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
MessageListenerContainer container) {
handle(thrownException, data);
}
/**
* Handle the exception.
* @param thrownException the exception.
* @param data the consumer records.
* @param consumer the consumer.
* @param container the container.
* @param invokeListener a callback to re-invoke the listener.
* @since 2.3.7
*/
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data,
Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
handle(thrownException, data);
}
}

8
spring-kafka/src/main/java/org/springframework/kafka/listener/CommonContainerStoppingErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 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.
@@ -70,12 +70,6 @@ public class CommonContainerStoppingErrorHandler extends KafkaExceptionLogLevelA
this.stopContainerAbnormally = stopContainerAbnormally;
}
@Override
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
public boolean remainingRecords() {
return true;
}
@Override
public boolean seeksAfterHandling() {
// We don't actually do any seeks here, but stopping the container has the same effect.

11
spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 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.
@@ -92,12 +92,6 @@ public class CommonDelegatingErrorHandler implements CommonErrorHandler {
this.causeChainTraversing = causeChainTraversing;
}
@SuppressWarnings("deprecation")
@Override
public boolean remainingRecords() {
return this.defaultErrorHandler.remainingRecords();
}
@Override
public boolean seeksAfterHandling() {
return this.defaultErrorHandler.seeksAfterHandling();
@@ -131,12 +125,9 @@ public class CommonDelegatingErrorHandler implements CommonErrorHandler {
@SuppressWarnings("deprecation")
private void checkDelegates() {
boolean remainingRecords = this.defaultErrorHandler.remainingRecords();
boolean ackAfterHandle = this.defaultErrorHandler.isAckAfterHandle();
boolean seeksAfterHandling = this.defaultErrorHandler.seeksAfterHandling();
this.delegates.values().forEach(handler -> {
Assert.isTrue(remainingRecords == handler.remainingRecords(),
"All delegates must return the same value when calling 'remainingRecords()'");
Assert.isTrue(ackAfterHandle == handler.isAckAfterHandle(),
"All delegates must return the same value when calling 'isAckAfterHandle()'");
Assert.isTrue(seeksAfterHandling == handler.seeksAfterHandling(),

72
spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 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.
@@ -25,6 +25,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.support.KafkaUtils;
import org.springframework.kafka.support.TopicPartitionOffset;
/**
@@ -37,29 +39,13 @@ import org.springframework.kafka.support.TopicPartitionOffset;
*/
public interface CommonErrorHandler extends DeliveryAttemptAware {
/**
* Return false (default) if this error handler should only receive the current failed
* record; remaining records will be passed to the listener after the error handler
* returns. When true, all remaining records including the failed record are passed to
* the error handler.
* @return false to receive only the failed record.
* @deprecated in favor of {@link #seeksAfterHandling()}.
* @see #handleRecord(Exception, ConsumerRecord, Consumer, MessageListenerContainer)
* @see #handleRemaining(Exception, List, Consumer, MessageListenerContainer)
*/
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
default boolean remainingRecords() {
return false;
}
/**
* Return true if this error handler performs seeks on the failed record and remaining
* records (or just the remaining records after a failed record is recovered).
* @return true if the next poll should fetch records.
*/
@SuppressWarnings("deprecation")
default boolean seeksAfterHandling() {
return remainingRecords();
return false;
}
/**
@@ -81,31 +67,11 @@ public interface CommonErrorHandler extends DeliveryAttemptAware {
default void handleOtherException(Exception thrownException, Consumer<?, ?> consumer,
MessageListenerContainer container, boolean batchListener) {
LogFactory.getLog(getClass()).error("'handleOtherException' is not implemented by this handler",
thrownException);
logger().error(thrownException, "'handleOtherException' is not implemented by this handler");
}
/**
* Handle the exception for a record listener when {@link #remainingRecords()} returns
* false. Use this to handle just the single failed record; remaining records from the
* poll will be sent to the listener.
* @param thrownException the exception.
* @param record the record.
* @param consumer the consumer.
* @param container the container.
* @deprecated in favor of
* {@link #handleOne(Exception, ConsumerRecord, Consumer, MessageListenerContainer)}.
* @see #remainingRecords()
*/
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
default void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
MessageListenerContainer container) {
LogFactory.getLog(getClass()).error("'handleRecord' is not implemented by this handler", thrownException);
}
/**
* Handle the exception for a record listener when {@link #remainingRecords()} returns
* Handle the exception for a record listener when {@link #seeksAfterHandling()} returns
* false. Use this to handle just the single failed record.
* @param thrownException the exception.
* @param record the record.
@@ -114,19 +80,14 @@ public interface CommonErrorHandler extends DeliveryAttemptAware {
* @return true if the error was "handled" or false if not and the container will
* re-submit the record to the listener.
* @since 2.9
* @see #remainingRecords()
* @see #seeksAfterHandling()
*/
@SuppressWarnings("deprecation")
default boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
MessageListenerContainer container) {
try {
handleRecord(thrownException, record, consumer, container);
return true;
}
catch (Exception ex) {
return false;
}
logger().error(thrownException, () -> "'handleOne' is not implemented by this handler for "
+ KafkaUtils.format(record));
return true;
}
/**
@@ -143,7 +104,7 @@ public interface CommonErrorHandler extends DeliveryAttemptAware {
default void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
MessageListenerContainer container) {
LogFactory.getLog(getClass()).error("'handleRemaining' is not implemented by this handler", thrownException);
logger().error(thrownException, "'handleRemaining' is not implemented by this handler");
}
/**
@@ -159,7 +120,16 @@ public interface CommonErrorHandler extends DeliveryAttemptAware {
default void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data,
Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
LogFactory.getLog(getClass()).error("'handleBatch' is not implemented by this handler", thrownException);
logger().error(thrownException, "'handleBatch' is not implemented by this handler");
}
/**
* Common error handler logger.
* @return the logger.
* @since 3.1
*/
default LogAccessor logger() {
return new LogAccessor(LogFactory.getLog(getClass()));
}
/**

8
spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 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.
@@ -51,12 +51,6 @@ public class CommonMixedErrorHandler implements CommonErrorHandler {
this.batchErrorHandler = batchErrorHandler;
}
@SuppressWarnings("deprecation")
@Override
public boolean remainingRecords() {
return this.recordErrorHandler.remainingRecords();
}
@Override
public boolean seeksAfterHandling() {
return this.recordErrorHandler.seeksAfterHandling();

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

@@ -272,7 +272,6 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
container.setApplicationEventPublisher(publisher);
}
container.setClientIdSuffix(this.concurrency > 1 || this.alwaysClientIdSuffix ? "-" + index : "");
container.setGenericErrorHandler(getGenericErrorHandler());
container.setCommonErrorHandler(getCommonErrorHandler());
container.setAfterRollbackProcessor(getAfterRollbackProcessor());
container.setRecordInterceptor(getRecordInterceptor());

52
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java
View File

@@ -1,52 +0,0 @@
/*
* Copyright 2017-2022 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
*
* https://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;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.lang.Nullable;
/**
* An error handler that has access to the consumer, for example to adjust
* offsets after an error.
*
* @author Gary Russell
* @since 2.0
* @deprecated in favor of {@link CommonErrorHandler}.
*
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface ConsumerAwareBatchErrorHandler extends BatchErrorHandler {
@Override
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data) {
throw new UnsupportedOperationException("Container should never call this");
}
@Override
void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer);
@Override
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
MessageListenerContainer container) {
handle(thrownException, data, consumer);
}
}

53
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareErrorHandler.java
View File

@@ -1,53 +0,0 @@
/*
* Copyright 2017-2022 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
*
* https://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;
import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.lang.Nullable;
/**
* An error handler that has access to the consumer, for example to adjust
* offsets after an error.
*
* @author Gary Russell
* @since 2.0
* @deprecated in favor of {@link CommonErrorHandler}.
*
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface ConsumerAwareErrorHandler extends ErrorHandler {
@Override
default void handle(Exception thrownException, @Nullable ConsumerRecord<?, ?> data) {
throw new UnsupportedOperationException("Container should never call this");
}
@Override
void handle(Exception thrownException, @Nullable ConsumerRecord<?, ?> data, Consumer<?, ?> consumer);
@Override
default void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> data, Consumer<?, ?> consumer,
MessageListenerContainer container) {
handle(thrownException, null, consumer);
}
}

36
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordInterceptor.java
View File

@@ -1,36 +0,0 @@
/*
* Copyright 2021-2022 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
*
* https://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;
import org.apache.kafka.clients.consumer.Consumer;
/**
* A {@link RecordInterceptor} that has access to the {@link Consumer}.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Gary Russell
* @since 2.7
* @deprecated - use {@link RecordInterceptor}.
*
*/
@Deprecated(since = "3.0", forRemoval = true) // in 3.1
@FunctionalInterface
public interface ConsumerAwareRecordInterceptor<K, V> extends RecordInterceptor<K, V> {
}

62
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java
View File

@@ -1,62 +0,0 @@
/*
* Copyright 2017-2022 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
*
* https://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;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.lang.Nullable;
/**
* An error handler that has access to the batch of records from the last poll the
* consumer, and the container.
*
* @author Gary Russell
* @since 2.1
* @deprecated in favor of {@link CommonErrorHandler}.
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface ContainerAwareBatchErrorHandler extends ConsumerAwareBatchErrorHandler {
@Override
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer) {
throw new UnsupportedOperationException("Container should never call this");
}
@Override
void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
MessageListenerContainer container);
/**
* Handle the exception.
* @param thrownException the exception.
* @param data the consumer records.
* @param consumer the consumer.
* @param container the container.
* @param invokeListener a callback to re-invoke the listener.
* @since 2.3.7
*/
@Override
@SuppressWarnings("unused")
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data,
Consumer<?, ?> consumer, MessageListenerContainer container, @Nullable Runnable invokeListener) {
handle(thrownException, data, consumer, container);
}
}

52
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java
View File

@@ -1,52 +0,0 @@
/*
* Copyright 2017-2022 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
*
* https://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;
import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.lang.Nullable;
/**
* An error handler that has access to the unprocessed records from the last poll
* (including the failed record), the consumer, and the container.
* The records passed to the handler will not be passed to the listener
* (unless re-fetched if the handler performs seeks).
*
* @author Gary Russell
* @since 2.1
* @deprecated in favor of {@link CommonErrorHandler}.
*
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface ContainerAwareErrorHandler extends RemainingRecordsErrorHandler {
@Override
default void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
Consumer<?, ?> consumer) {
throw new UnsupportedOperationException("Container should never call this");
}
@Override
void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
MessageListenerContainer container);
}

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

@@ -867,7 +867,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
* @see #setHeaderNamesSupplier(Supplier)
*/
@Nullable
@Deprecated(since = "3.0.9", forRemoval = true)
@Deprecated(since = "3.0.9", forRemoval = true) // 3.2
protected HeaderNames getHeaderNames() {
return null;
}

8
spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 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.
@@ -133,12 +133,6 @@ public class DefaultErrorHandler extends FailedBatchProcessor implements CommonE
this.ackAfterHandle = ackAfterHandle;
}
@Override
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
public boolean remainingRecords() {
return isSeekAfterError();
}
@Override
public boolean seeksAfterHandling() {
return isSeekAfterError();

46
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java
View File

@@ -1,46 +0,0 @@
/*
* Copyright 2015-2022 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
*
* https://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;
import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
/**
* Handles errors thrown during the execution of a {@link MessageListener}.
*
* @author Marius Bogoevici
* @author Gary Russell
* @deprecated in favor of {@link CommonErrorHandler}.
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
public interface ErrorHandler extends GenericErrorHandler<ConsumerRecord<?, ?>> {
/**
* Handle the exception.
* @param thrownException the exception.
* @param records the remaining records including the one that failed.
* @param consumer the consumer.
* @param container the container.
*/
default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
MessageListenerContainer container) {
handle(thrownException, null);
}
}

176
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java
View File

@@ -1,176 +0,0 @@
/*
* Copyright 2021-2022 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
*
* https://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;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.util.Assert;
/**
* Adapts a legacy {@link ErrorHandler} or {@link BatchErrorHandler}.
*
* @author Gary Russell
* @since 2.7.4
*
*/
@SuppressWarnings("deprecation")
class ErrorHandlerAdapter implements CommonErrorHandler {
@SuppressWarnings({ "rawtypes", "unchecked" })
private static final ConsumerRecords EMPTY_BATCH = new ConsumerRecords(Collections.emptyMap());
private final ErrorHandler errorHandler;
private final BatchErrorHandler batchErrorHandler;
/**
* Adapt an {@link ErrorHandler}.
* @param errorHandler the handler.
*/
ErrorHandlerAdapter(ErrorHandler errorHandler) {
Assert.notNull(errorHandler, "'errorHandler' cannot be null");
this.errorHandler = errorHandler;
this.batchErrorHandler = null;
}
/**
* Adapt a {@link BatchErrorHandler}.
* @param batchErrorHandler the handler.
*/
ErrorHandlerAdapter(BatchErrorHandler batchErrorHandler) {
Assert.notNull(batchErrorHandler, "'batchErrorHandler' cannot be null");
this.errorHandler = null;
this.batchErrorHandler = batchErrorHandler;
}
@SuppressWarnings("deprecation")
@Override
public boolean remainingRecords() {
return this.errorHandler instanceof RemainingRecordsErrorHandler;
}
@Override
public boolean deliveryAttemptHeader() {
return this.errorHandler instanceof DeliveryAttemptAware;
}
@Override
public void clearThreadState() {
if (this.errorHandler != null) {
this.errorHandler.clearThreadState();
}
else {
this.batchErrorHandler.clearThreadState();
}
}
@Override
public boolean isAckAfterHandle() {
if (this.errorHandler != null) {
return this.errorHandler.isAckAfterHandle();
}
else {
return this.batchErrorHandler.isAckAfterHandle();
}
}
@Override
public void setAckAfterHandle(boolean ack) {
if (this.errorHandler != null) {
this.errorHandler.setAckAfterHandle(ack);
}
else {
this.batchErrorHandler.setAckAfterHandle(ack);
}
}
@Override
public int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) {
Assert.state(deliveryAttemptHeader(), "This method should not be called by the container");
return ((DeliveryAttemptAware) this.errorHandler).deliveryAttempt(topicPartitionOffset);
}
@SuppressWarnings({ "unchecked" })
@Override
public void handleOtherException(Exception thrownException, Consumer<?, ?> consumer,
MessageListenerContainer container, boolean batchListener) {
if (this.errorHandler != null) {
this.errorHandler.handle(thrownException, Collections.emptyList(), consumer, container);
}
else {
this.batchErrorHandler.handle(thrownException, EMPTY_BATCH, consumer, container, () -> { });
}
}
@SuppressWarnings("deprecation")
@Override
public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
MessageListenerContainer container) {
if (this.errorHandler != null) {
this.errorHandler.handle(thrownException, record, consumer);
}
else {
CommonErrorHandler.super.handleRecord(thrownException, record, consumer, container);
}
}
@Override
public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
MessageListenerContainer container) {
if (this.errorHandler != null) {
this.errorHandler.handle(thrownException, records, consumer, container);
}
else {
CommonErrorHandler.super.handleRemaining(thrownException, records, consumer, container);
}
}
@Override
public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
MessageListenerContainer container, Runnable invokeListener) {
if (this.batchErrorHandler != null) {
this.batchErrorHandler.handle(thrownException, data, consumer, container, invokeListener);
}
else {
CommonErrorHandler.super.handleBatch(thrownException, data, consumer, container, invokeListener);
}
}
@Override
public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions,
Runnable publishPause) {
if (this.batchErrorHandler instanceof FallbackBatchErrorHandler) {
((FallbackBatchErrorHandler) this.batchErrorHandler).onPartitionsAssigned(consumer, partitions,
publishPause);
}
}
}

30
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java
View File

@@ -57,36 +57,6 @@ public final class ErrorHandlingUtils {
private ErrorHandlingUtils() {
}
/**
* Retry a complete batch by pausing the consumer and then, in a loop, poll the
* consumer, wait for the next back off, then call the listener. When retries are
* exhausted, call the recoverer with the {@link ConsumerRecords}.
* @param thrownException the exception.
* @param records the records.
* @param consumer the consumer.
* @param container the container.
* @param invokeListener the {@link Runnable} to run (call the listener).
* @param backOff the backOff.
* @param seeker the common error handler that re-seeks the entire batch.
* @param recoverer the recoverer.
* @param logger the logger.
* @param logLevel the log level.
* @param retryListeners the retry listeners.
* @param classifier the exception classifier.
* @since 2.8.11
* @deprecated in favor of
* {@link #retryBatch(Exception, ConsumerRecords, Consumer, MessageListenerContainer, Runnable, BackOff, CommonErrorHandler, BiConsumer, LogAccessor, org.springframework.kafka.KafkaException.Level, List, BinaryExceptionClassifier, boolean)}.
*/
@Deprecated
public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
MessageListenerContainer container, Runnable invokeListener, BackOff backOff,
CommonErrorHandler seeker, BiConsumer<ConsumerRecords<?, ?>, Exception> recoverer, LogAccessor logger,
KafkaException.Level logLevel, List<RetryListener> retryListeners, BinaryExceptionClassifier classifier) {
retryBatch(thrownException, records, consumer, container, invokeListener, backOff, seeker, recoverer, logger,
logLevel, retryListeners, classifier, false);
}
/**
* Retry a complete batch by pausing the consumer and then, in a loop, poll the
* consumer, wait for the next back off, then call the listener. When retries are

82
spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java
View File

@@ -1,82 +0,0 @@
/*
* Copyright 2016-2022 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
*
* https://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;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.lang.Nullable;
/**
* A generic error handler.
*
* @param <T> the data type.
*
* @author Gary Russell
* @since 1.1
* @deprecated in favor of {@link CommonErrorHandler}.
*
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface GenericErrorHandler<T> {
/**
* Handle the exception.
* @param thrownException The exception.
* @param data the data.
*/
void handle(Exception thrownException, @Nullable T data);
/**
* Handle the exception.
* @param thrownException The exception.
* @param data the data.
* @param consumer the consumer.
*/
default void handle(Exception thrownException, @Nullable T data, Consumer<?, ?> consumer) {
handle(thrownException, data);
}
/**
* Optional method to clear thread state; will be called just before a consumer
* thread terminates.
* @since 2.3
*/
default void clearThreadState() {
}
/**
* Return true if the offset should be committed for a handled error (no exception
* thrown).
* @return true to commit.
* @since 2.3.2
*/
default boolean isAckAfterHandle() {
return true;
}
/**
* Set to false to prevent the container from committing the offset of a recovered
* record (when the error handler does not itself throw an exception).
* @param ack false to not commit.
* @since 2.5.6
*/
default void setAckAfterHandle(boolean ack) {
throw new UnsupportedOperationException("This error handler does not support setting this property");
}
}

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

@@ -1012,40 +1012,13 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
return null;
}
@SuppressWarnings("deprecation")
@Nullable
private CommonErrorHandler determineCommonErrorHandler() {
CommonErrorHandler common = getCommonErrorHandler();
GenericErrorHandler<?> errHandler = getGenericErrorHandler();
if (common != null) {
if (errHandler != null) {
this.logger.debug("GenericErrorHandler is ignored when a CommonErrorHandler is provided");
}
return common;
}
if (errHandler == null && this.transactionManager == null) {
return new DefaultErrorHandler();
}
if (this.isBatchListener) {
validateErrorHandler(true, errHandler);
BatchErrorHandler batchErrorHandler = (BatchErrorHandler) errHandler;
if (batchErrorHandler != null) {
return new ErrorHandlerAdapter(batchErrorHandler);
}
else {
return null;
}
}
else {
validateErrorHandler(false, errHandler);
ErrorHandler eh = (ErrorHandler) errHandler;
if (eh != null) {
return new ErrorHandlerAdapter(eh);
}
else {
return null;
}
if (common == null && this.transactionManager == null) {
common = new DefaultErrorHandler();
}
return common;
}
String getClientId() {
@@ -1315,19 +1288,6 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
}
@SuppressWarnings("deprecation")
private void validateErrorHandler(boolean batch, @Nullable GenericErrorHandler<?> errHandler) {
if (errHandler == null) {
return;
}
Class<?> clazz = errHandler.getClass();
Assert.state(batch
? BatchErrorHandler.class.isAssignableFrom(clazz)
: ErrorHandler.class.isAssignableFrom(clazz),
() -> "Error handler is not compatible with the message listener, expecting an instance of "
+ (batch ? "BatchErrorHandler" : "ErrorHandler") + " not " + errHandler.getClass().getName());
}
@Override
public boolean isLongLived() {
return true;
@@ -2493,7 +2453,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
ConsumerRecords<K, V> afterHandling = this.commonErrorHandler.handleBatchAndReturnRemaining(rte,
records, this.consumer, KafkaMessageListenerContainer.this.thisOrParentContainer,
() -> invokeBatchOnMessageWithRecordsOrList(records, list));
if (!afterHandling.isEmpty()) {
if (afterHandling != null && !afterHandling.isEmpty()) {
this.remainingRecords = afterHandling;
this.pauseForPending = true;
}

47
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerInvokingBatchErrorHandler.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright 2020-2022 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
*
* https://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;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.lang.Nullable;
/**
* A batch error handler that is capable of invoking the listener during error handling.
*
* @author Gary Russell
* @since 2.3.7
* @deprecated in favor of {@link CommonErrorHandler}.
*
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface ListenerInvokingBatchErrorHandler extends ContainerAwareBatchErrorHandler {
@Override
default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
MessageListenerContainer container) {
throw new UnsupportedOperationException("Container should never call this");
}
@Override
void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records,
Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener);
}

69
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java
View File

@@ -16,20 +16,11 @@
package org.springframework.kafka.listener;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectStreamClass;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.kafka.support.serializer.SerializationUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.BackOffExecution;
@@ -83,64 +74,6 @@ public final class ListenerUtils {
return listenerType;
}
/**
* Extract a {@link DeserializationException} from the supplied header name, if
* present.
* @param record the consumer record.
* @param headerName the header name.
* @param logger the logger for logging errors.
* @return the exception or null.
* @since 2.3
* @deprecated in favor of
* {@link SerializationUtils#getExceptionFromHeader(ConsumerRecord, String, LogAccessor)}.
*/
@Deprecated
@Nullable
public static DeserializationException getExceptionFromHeader(final ConsumerRecord<?, ?> record,
String headerName, LogAccessor logger) {
return SerializationUtils.getExceptionFromHeader(record, headerName, logger);
}
/**
* Convert a byte array containing a serialized {@link DeserializationException} to the
* {@link DeserializationException}.
* @param logger a log accessor to log errors.
* @param value the bytes.
* @return the exception or null if deserialization fails.
* @since 2.8.1
* @deprecated in favor of
* {@link SerializationUtils#getExceptionFromHeader(ConsumerRecord, String, LogAccessor)} or
* {@link SerializationUtils#byteArrayToDeserializationException(LogAccessor, org.apache.kafka.common.header.Header)}.
*/
@Deprecated
@Nullable
public static DeserializationException byteArrayToDeserializationException(LogAccessor logger, byte[] value) {
try {
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(value)) {
boolean first = true;
@Override
protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {
if (this.first) {
this.first = false;
Assert.state(desc.getName().equals(DeserializationException.class.getName()),
"Header does not contain a DeserializationException");
}
return super.resolveClass(desc);
}
};
return (DeserializationException) ois.readObject();
}
catch (IOException | ClassNotFoundException | ClassCastException e) {
logger.error(e, "Failed to deserialize a deserialization exception");
return null;
}
}
/**
* Sleep according to the {@link BackOff}; when the {@link BackOffExecution} returns
* {@link BackOffExecution#STOP} sleep for the previous backOff.
@@ -155,7 +88,7 @@ public final class ListenerUtils {
* @deprecated in favor of
* {@link #unrecoverableBackOff(BackOff, Map, Map, MessageListenerContainer)}.
*/
@Deprecated
@Deprecated(since = "3.1", forRemoval = true) // 3.2
public static void unrecoverableBackOff(BackOff backOff, ThreadLocal<BackOffExecution> executions,
ThreadLocal<Long> lastIntervals, MessageListenerContainer container) throws InterruptedException {

61
spring-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java
View File

@@ -1,61 +0,0 @@
/*
* Copyright 2017-2022 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
*
* https://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;
import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.lang.Nullable;
/**
* An error handler that has access to the unprocessed records from the last poll
* (including the failed record) and the consumer, for example to adjust offsets after an
* error. The records passed to the handler will not be passed to the listener
* (unless re-fetched if the handler performs seeks).
*
* @author Gary Russell
* @since 2.0.1
* @deprecated in favor of {@link CommonErrorHandler}.
*
*/
@Deprecated(since = "2.8", forRemoval = true) // in 3.1
@FunctionalInterface
public interface RemainingRecordsErrorHandler extends ConsumerAwareErrorHandler {
@Override
default void handle(Exception thrownException, @Nullable ConsumerRecord<?, ?> data, Consumer<?, ?> consumer) {
throw new UnsupportedOperationException("Container should never call this");
}
/**
* Handle the exception. The failed record is the first in the list.
* @param thrownException the exception.
* @param records the remaining records including the one that failed.
* @param consumer the consumer.
*/
void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer);
@Override
default void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
Consumer<?, ?> consumer, MessageListenerContainer container) {
handle(thrownException, records, consumer);
}
}

9
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java
View File

@@ -68,12 +68,6 @@ public class DefaultDestinationTopicResolver extends ExceptionClassifier
private boolean contextRefreshed;
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
public DefaultDestinationTopicResolver(Clock clock, ApplicationContext applicationContext) {
this(clock);
this.applicationContext = applicationContext;
}
/**
* Constructs an instance with the given clock.
* @param clock the clock to be used for time-based operations
@@ -135,8 +129,7 @@ public class DefaultDestinationTopicResolver extends ExceptionClassifier
@SuppressWarnings("deprecation")
private DestinationTopic resolveRetryDestination(DestinationTopicHolder destinationTopicHolder) {
return ((destinationTopicHolder.getSourceDestination().isReusableRetryTopic()) ||
(destinationTopicHolder.getSourceDestination().isSingleTopicRetry()))
return (destinationTopicHolder.getSourceDestination().isReusableRetryTopic())
? destinationTopicHolder.getSourceDestination()
: destinationTopicHolder.getNextDestination();
}

23
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java
View File

@@ -72,18 +72,6 @@ public class DestinationTopic {
return Type.REUSABLE_RETRY_TOPIC.equals(this.properties.type);
}
/**
* Whether this is a single retry topic.
*
* @return whether this is a single retry topic.
* @deprecated in favor of using {@link DestinationTopic.Type#REUSABLE_RETRY_TOPIC}
* and {@link #isReusableRetryTopic()}.
*/
@Deprecated
public boolean isSingleTopicRetry() {
return Type.SINGLE_TOPIC_RETRY.equals(this.properties.type);
}
public boolean isMainTopic() {
return Type.MAIN.equals(this.properties.type);
}
@@ -225,8 +213,7 @@ public class DestinationTopic {
}
public boolean isRetryTopic() {
return Type.RETRY.equals(this.type) || Type.SINGLE_TOPIC_RETRY.equals(this.type)
|| Type.REUSABLE_RETRY_TOPIC.equals(this.type);
return Type.RETRY.equals(this.type) || Type.REUSABLE_RETRY_TOPIC.equals(this.type);
}
public String suffix() {
@@ -301,14 +288,6 @@ public class DestinationTopic {
RETRY,
/**
* A single retry topic for all retries.
*
* @deprecated Use {@code REUSABLE_RETRY_TOPIC} instead.
*/
@Deprecated
SINGLE_TOPIC_RETRY,
/**
* A retry topic reused along sequential retries
* with the same backoff interval.

53
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java
View File

@@ -41,8 +41,6 @@ import org.springframework.util.StringUtils;
*/
public class DestinationTopicPropertiesFactory {
private static final String DEPRECATION = "deprecation";
private static final String MAIN_TOPIC_SUFFIX = "";
private final DestinationTopicSuffixes destinationTopicSuffixes;
@@ -57,9 +55,6 @@ public class DestinationTopicPropertiesFactory {
private final KafkaOperations<?, ?> kafkaOperations;
@SuppressWarnings(DEPRECATION)
private final FixedDelayStrategy fixedDelayStrategy;
private final DltStrategy dltStrategy;
private final TopicSuffixingStrategy topicSuffixingStrategy;
@@ -71,11 +66,23 @@ public class DestinationTopicPropertiesFactory {
@Nullable
private Boolean autoStartDltHandler;
@SuppressWarnings(DEPRECATION)
/**
* Construct an instance with the provided properties.
* @param retryTopicSuffix the suffix.
* @param dltSuffix the dlt suffix.
* @param backOffValues the back off values.
* @param exceptionClassifier the exception classifier.
* @param numPartitions the number of partitions.
* @param kafkaOperations the operations.
* @param dltStrategy the dlt strategy.
* @param topicSuffixingStrategy the topic suffixing strategy.
* @param sameIntervalTopicReuseStrategy the same interval reuse strategy.
* @param timeout the timeout.
* @since 3.0.12
*/
public DestinationTopicPropertiesFactory(String retryTopicSuffix, String dltSuffix, List<Long> backOffValues,
BinaryExceptionClassifier exceptionClassifier,
int numPartitions, KafkaOperations<?, ?> kafkaOperations,
FixedDelayStrategy fixedDelayStrategy,
DltStrategy dltStrategy,
TopicSuffixingStrategy topicSuffixingStrategy,
SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy,
@@ -85,7 +92,6 @@ public class DestinationTopicPropertiesFactory {
this.kafkaOperations = kafkaOperations;
this.exceptionClassifier = exceptionClassifier;
this.numPartitions = numPartitions;
this.fixedDelayStrategy = fixedDelayStrategy;
this.topicSuffixingStrategy = topicSuffixingStrategy;
this.sameIntervalTopicReuseStrategy = sameIntervalTopicReuseStrategy;
this.timeout = timeout;
@@ -95,18 +101,6 @@ public class DestinationTopicPropertiesFactory {
this.maxAttempts = this.backOffValues.size() + 1;
}
@SuppressWarnings(DEPRECATION)
public DestinationTopicPropertiesFactory(String retryTopicSuffix, String dltSuffix, List<Long> backOffValues,
BinaryExceptionClassifier exceptionClassifier,
int numPartitions, KafkaOperations<?, ?> kafkaOperations,
FixedDelayStrategy fixedDelayStrategy,
DltStrategy dltStrategy,
TopicSuffixingStrategy topicSuffixingStrategy,
long timeout) {
this(retryTopicSuffix, dltSuffix, backOffValues, exceptionClassifier, numPartitions, kafkaOperations,
fixedDelayStrategy, dltStrategy, topicSuffixingStrategy, SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS,
timeout);
}
/**
* Set to false to not start the DLT handler.
* @param autoStart false to not start.
@@ -134,12 +128,7 @@ public class DestinationTopicPropertiesFactory {
}
private boolean isSingleTopicFixedDelay() {
return isFixedDelay() && (isSingleTopicStrategy() || isSingleTopicSameIntervalTopicReuseStrategy());
}
@SuppressWarnings(DEPRECATION)
private boolean isSingleTopicStrategy() {
return FixedDelayStrategy.SINGLE_TOPIC.equals(this.fixedDelayStrategy);
return isFixedDelay() && isSingleTopicSameIntervalTopicReuseStrategy();
}
private boolean isSingleTopicSameIntervalTopicReuseStrategy() {
@@ -169,9 +158,7 @@ public class DestinationTopicPropertiesFactory {
// the end of the list.
? amountOfDuplicates(this.backOffValues.get(this.backOffValues.size() - 1)) - 1
: 0
: isSingleTopicStrategy()
? this.backOffValues.size() - 1
: 0
: 0
: 0;
}
@@ -203,14 +190,12 @@ public class DestinationTopicPropertiesFactory {
return (attempt, throwable) -> attempt < this.maxAttempts && this.exceptionClassifier.classify(throwable);
}
@SuppressWarnings(DEPRECATION)
private DestinationTopic.Properties createRetryProperties(int index,
BiPredicate<Integer, Throwable> shouldRetryOn) {
BiPredicate<Integer, Throwable> shouldRetryOn) {
int indexInBackoffValues = index - 1;
Long thisBackOffValue = this.backOffValues.get(indexInBackoffValues);
DestinationTopic.Type topicTypeToUse = isSingleTopicFixedDelay()
? Type.SINGLE_TOPIC_RETRY
: isDelayWithReusedTopic(thisBackOffValue)
DestinationTopic.Type topicTypeToUse = isDelayWithReusedTopic(thisBackOffValue)
? Type.REUSABLE_RETRY_TOPIC
: Type.RETRY;
return createProperties(topicTypeToUse, shouldRetryOn, indexInBackoffValues,

41
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/FixedDelayStrategy.java
View File

@@ -1,41 +0,0 @@
/*
* Copyright 2018-2023 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
*
* https://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.retrytopic;
/**
*
* Defines the topic strategy to handle fixed delays.
*
* @author Tomaz Fernandes
* @since 2.7
* @deprecated in favor of {@link SameIntervalTopicReuseStrategy}.
*
*/
@Deprecated
public enum FixedDelayStrategy {
/**
* Uses a single topic to achieve non-blocking retry.
*/
SINGLE_TOPIC,
/**
* Uses one separate topic per retry attempt.
*/
MULTIPLE_TOPICS
}

31
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java
View File

@@ -71,9 +71,6 @@ public class RetryTopicConfigurationBuilder {
@Nullable
private BinaryExceptionClassifierBuilder classifierBuilder;
@SuppressWarnings("deprecation")
private FixedDelayStrategy fixedDelayStrategy = FixedDelayStrategy.MULTIPLE_TOPICS;
private DltStrategy dltStrategy = DltStrategy.ALWAYS_RETRY_ON_ERROR;
private long timeout = RetryTopicConstants.NOT_SET;
@@ -406,32 +403,6 @@ public class RetryTopicConfigurationBuilder {
return this;
}
/**
* Configure the use of a single retry topic with fixed delays.
* @return the builder.
* @deprecated in favor of {@link #useSingleTopicForSameIntervals()}.
* @see FixedDelayStrategy#SINGLE_TOPIC
*/
@Deprecated
public RetryTopicConfigurationBuilder useSingleTopicForFixedDelays() {
this.fixedDelayStrategy = FixedDelayStrategy.SINGLE_TOPIC;
return this;
}
/**
* Configure the {@link FixedDelayStrategy}; default is
* {@link FixedDelayStrategy#MULTIPLE_TOPICS}.
* @param delayStrategy the delay strategy.
* @return the builder.
* @deprecated in favor of
* {@link #sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy)}.
*/
@Deprecated
public RetryTopicConfigurationBuilder useSingleTopicForFixedDelays(FixedDelayStrategy delayStrategy) {
this.fixedDelayStrategy = delayStrategy;
return this;
}
/* ---------------- Configure Topics Auto Creation -------------- */
/**
@@ -592,7 +563,7 @@ public class RetryTopicConfigurationBuilder {
List<DestinationTopic.Properties> destinationTopicProperties =
new DestinationTopicPropertiesFactory(this.retryTopicSuffix, this.dltSuffix, backOffValues,
buildClassifier(), this.topicCreationConfiguration.getNumPartitions(),
sendToTopicKafkaTemplate, this.fixedDelayStrategy, this.dltStrategy,
sendToTopicKafkaTemplate, this.dltStrategy,
this.topicSuffixingStrategy, this.sameIntervalTopicReuseStrategy, this.timeout)
.autoStartDltHandler(this.autoStartDltHandler)
.createProperties();

20
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java
View File

@@ -240,26 +240,6 @@ public class RetryTopicConfigurer implements BeanFactoryAware {
private final RetryTopicNamesProviderFactory retryTopicNamesProviderFactory;
/**
* Create an instance with the provided properties.
* @param destinationTopicProcessor the destination topic processor.
* @param containerFactoryResolver the container factory resolver.
* @param listenerContainerFactoryConfigurer the container factory configurer.
* @param beanFactory the bean factory.
* @param retryTopicNamesProviderFactory the retry topic names factory.
*/
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
public RetryTopicConfigurer(DestinationTopicProcessor destinationTopicProcessor,
ListenerContainerFactoryResolver containerFactoryResolver,
ListenerContainerFactoryConfigurer listenerContainerFactoryConfigurer,
BeanFactory beanFactory,
RetryTopicNamesProviderFactory retryTopicNamesProviderFactory) {
this(destinationTopicProcessor, containerFactoryResolver,
listenerContainerFactoryConfigurer, retryTopicNamesProviderFactory);
this.beanFactory = beanFactory;
}
/**
* Create an instance with the provided properties.
* @param destinationTopicProcessor the destination topic processor.

115
spring-kafka/src/main/java/org/springframework/kafka/streams/HeaderEnricher.java
View File

@@ -1,115 +0,0 @@
/*
* Copyright 2019 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
*
* https://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.streams;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.springframework.expression.Expression;
/**
* Manipulate the headers.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Gary Russell
* @since 2.3
* @deprecated in favor of {@link HeaderEnricherProcessor}.
*
*/
@Deprecated
public class HeaderEnricher<K, V> implements Transformer<K, V, KeyValue<K, V>> {
private final Map<String, Expression> headerExpressions = new HashMap<>();
private ProcessorContext processorContext;
public HeaderEnricher(Map<String, Expression> headerExpressions) {
this.headerExpressions.putAll(headerExpressions);
}
@Override
public void init(ProcessorContext context) {
this.processorContext = context;
}
@Override
public KeyValue<K, V> transform(K key, V value) {
Headers headers = this.processorContext.headers();
Container<K, V> container = new Container<>(this.processorContext, key, value);
this.headerExpressions.forEach((name, expression) -> {
Object headerValue = expression.getValue(container);
if (headerValue instanceof String) {
headerValue = ((String) headerValue).getBytes(StandardCharsets.UTF_8);
}
else if (!(headerValue instanceof byte[])) {
throw new IllegalStateException("Invalid header value type: " + headerValue.getClass());
}
headers.add(new RecordHeader(name, (byte[]) headerValue));
});
return new KeyValue<>(key, value);
}
@Override
public void close() {
// NO-OP
}
/**
* Container object for SpEL evaluation.
*
* @param <K> the key type.
* @param <V> the value type.
*
*/
public static final class Container<K, V> {
private final ProcessorContext context;
private final K key;
private final V value;
private Container(ProcessorContext context, K key, V value) {
this.context = context;
this.key = key;
this.value = value;
}
public ProcessorContext getContext() {
return this.context;
}
public K getKey() {
return this.key;
}
public V getValue() {
return this.value;
}
}
}

106
spring-kafka/src/main/java/org/springframework/kafka/streams/messaging/MessagingTransformer.java
View File

@@ -1,106 +0,0 @@
/*
* Copyright 2019-2022 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
*
* https://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.streams.messaging;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* A {@link Transformer} implementation that invokes a {@link MessagingFunction}
* converting to/from spring-messaging {@link Message}. Can be used, for example,
* to invoke a Spring Integration flow.
*
* @param <K> the key type.
* @param <V> the value type.
* @param <R> the result value type.
*
* @author Gary Russell
* @since 2.3
* @deprecated in favor of {@link MessagingProcessor}.
*
*/
@Deprecated
public class MessagingTransformer<K, V, R> implements Transformer<K, V, KeyValue<K, R>> {
private final MessagingFunction function;
private final MessagingMessageConverter converter;
private ProcessorContext processorContext;
/**
* Construct an instance with the provided function and converter.
* @param function the function.
* @param converter the converter.
*/
public MessagingTransformer(MessagingFunction function, MessagingMessageConverter converter) {
Assert.notNull(function, "'function' cannot be null");
Assert.notNull(converter, "'converter' cannot be null");
this.function = function;
this.converter = converter;
}
@Override
public void init(ProcessorContext context) {
this.processorContext = context;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public KeyValue<K, R> transform(K key, V value) {
Headers headers = this.processorContext.headers();
ConsumerRecord<Object, Object> record = new ConsumerRecord<Object, Object>(this.processorContext.topic(),
this.processorContext.partition(), this.processorContext.offset(),
this.processorContext.timestamp(), TimestampType.NO_TIMESTAMP_TYPE,
0, 0,
key, value,
headers, Optional.empty());
Message<?> message = this.converter.toMessage(record, null, null, null);
message = this.function.exchange(message);
List<String> headerList = new ArrayList<>();
headers.forEach(header -> headerList.add(header.key()));
headerList.forEach(name -> headers.remove(name));
ProducerRecord<?, ?> fromMessage = this.converter.fromMessage(message, "dummy");
fromMessage.headers().forEach(header -> {
if (!header.key().equals(KafkaHeaders.TOPIC)) {
headers.add(header);
}
});
Object key2 = message.getHeaders().get(KafkaHeaders.KEY);
return new KeyValue(key2 == null ? key : key2, message.getPayload());
}
@Override
public void close() {
// NO-OP
}
}

2
spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java
View File

@@ -152,7 +152,7 @@ public final class KafkaUtils {
* @since 2.7.12
* @deprecated - no longer used.
*/
@Deprecated
@Deprecated(since = "3.1", forRemoval = true) // 3.2
public static void setLogOnlyMetadata(boolean onlyMeta) {
}

17
spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java
View File

@@ -53,23 +53,6 @@ public final class MicrometerHolder {
private final Function<Object, Map<String, String>> tagsProvider;
/**
* Create an instance with the provided properties.
* @param context the application context from which to obtain the meter registry.
* @param name the value of the 'name' tag.
* @param timerName the timer name.
* @param timerDesc the timer description.
* @param tags additional tags.
* @deprecated in favor of
* {@link #MicrometerHolder(ApplicationContext, String, String, String, Function)}.
*/
@Deprecated
public MicrometerHolder(@Nullable ApplicationContext context, String name,
String timerName, String timerDesc, Map<String, String> tags) {
this(context, name, timerName, timerDesc, cr -> tags);
}
/**
* Create an instance with the provided properties.
* @param context the application context from which to obtain the meter registry.

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

@@ -1136,12 +1136,13 @@ public class EnableKafkaIntegrationTests {
factory.setCommonErrorHandler(new CommonErrorHandler() {
@Override
public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record,
public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
Consumer<?, ?> consumer, MessageListenerContainer container) {
globalErrorThrowable = thrownException;
consumer.seek(new org.apache.kafka.common.TopicPartition(record.topic(), record.partition()),
record.offset());
return false;
}
});
factory.getContainerProperties().setMicrometerTags(Collections.singletonMap("extraTag", "foo"));

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java
View File

@@ -156,7 +156,7 @@ public class ConcurrentMessageListenerContainerMockTests {
container.setCommonErrorHandler(new CommonErrorHandler() {
@Override
public boolean remainingRecords() {
public boolean seeksAfterHandling() {
return true;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -625,10 +625,11 @@ public class ConcurrentMessageListenerContainerTests {
container.setCommonErrorHandler(new CommonErrorHandler() {
@Override
public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
MessageListenerContainer container) {
catchError.set(true);
return true;
}
});
container.start();

3
spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java
View File

@@ -185,7 +185,7 @@ public class ErrorHandlingDeserializerTests {
factory.setCommonErrorHandler(new CommonErrorHandler() {
@Override
public void handleRecord(Exception t, ConsumerRecord<?, ?> r,
public boolean handleOne(Exception t, ConsumerRecord<?, ?> r,
Consumer<?, ?> consumer, MessageListenerContainer container) {
if (r.value() == null && t.getCause() instanceof DeserializationException) {
@@ -196,6 +196,7 @@ public class ErrorHandlingDeserializerTests {
keyErrorCount.incrementAndGet();
}
latch.countDown();
return true;
}
});

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

@@ -1479,16 +1479,17 @@ public class KafkaMessageListenerContainerTests {
containerProps.setClientId("clientId");
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
BatchErrorHandler errorHandler = mock(BatchErrorHandler.class);
CommonErrorHandler errorHandler = mock(CommonErrorHandler.class);
given(errorHandler.isAckAfterHandle()).willReturn(true);
container.setBatchErrorHandler(errorHandler);
given(errorHandler.seeksAfterHandling()).willReturn(true);
container.setCommonErrorHandler(errorHandler);
container.start();
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
InOrder inOrder = inOrder(messageListener, consumer, errorHandler);
inOrder.verify(consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
inOrder.verify(messageListener).onMessage(any());
inOrder.verify(errorHandler).handle(any(), any(), any(), any(), any());
inOrder.verify(errorHandler).handleBatch(any(), any(), any(), any(), any());
inOrder.verify(consumer).commitSync(anyMap(), any());
container.stop();
}
@@ -2407,42 +2408,6 @@ public class KafkaMessageListenerContainerTests {
}
@Test
@SuppressWarnings("deprecation")
public void testBadErrorHandler() {
Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("foo");
containerProps.setMissingTopicsFatal(false);
KafkaMessageListenerContainer<Integer, Foo1> badContainer =
new KafkaMessageListenerContainer<>(cf, containerProps);
badContainer.setBatchErrorHandler((thrownException, data) -> {
});
badContainer.setupMessageListener((MessageListener<String, String>) m -> {
});
assertThatIllegalStateException().isThrownBy(() -> badContainer.start())
.withMessageContaining("Error handler is not compatible with the message listener");
}
@Test
@SuppressWarnings("deprecation")
public void testBadBatchErrorHandler() {
Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("foo");
containerProps.setMissingTopicsFatal(false);
KafkaMessageListenerContainer<Integer, Foo1> badContainer =
new KafkaMessageListenerContainer<>(cf, containerProps);
badContainer.setErrorHandler((thrownException, data) -> {
});
badContainer.setupMessageListener((BatchMessageListener<String, String>) m -> {
});
assertThatIllegalStateException().isThrownBy(() -> badContainer.start())
.withMessageContaining("Error handler is not compatible with the message listener");
}
@Test
public void testRebalanceAfterFailedRecord() throws Exception {
logger.info("Start rebalance after failed record");

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 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.
@@ -202,7 +202,7 @@ public class RemainingRecordsErrorHandlerTests {
factory.setCommonErrorHandler(new CommonErrorHandler() {
@Override
public boolean remainingRecords() {
public boolean seeksAfterHandling() {
return true;
}

72
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactoryTests.java
View File

@@ -51,10 +51,6 @@ class DestinationTopicPropertiesFactoryTests {
private final int numPartitions = 0;
@SuppressWarnings("deprecation")
private final FixedDelayStrategy fixedDelayStrategy =
FixedDelayStrategy.SINGLE_TOPIC;
private final TopicSuffixingStrategy suffixWithDelayValueSuffixingStrategy =
TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE;
@@ -94,7 +90,7 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations, fixedDelayStrategy,
classifier, numPartitions, kafkaOperations,
dltStrategy, suffixWithDelayValueSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy,
RetryTopicConstants.NOT_SET).createProperties();
@@ -140,7 +136,7 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations, fixedDelayStrategy,
classifier, numPartitions, kafkaOperations,
dltStrategy, TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE,
multipleTopicsSameIntervalReuseStrategy, RetryTopicConstants.NOT_SET).createProperties();
@@ -156,7 +152,6 @@ class DestinationTopicPropertiesFactoryTests {
assertThat(firstRetryProperties.isDltTopic()).isFalse();
assertThat(firstRetryProperties.isRetryTopic()).isTrue();
DestinationTopic firstRetryDestinationTopic = destinationTopicList.get(1);
assertThat(firstRetryDestinationTopic.isSingleTopicRetry()).isFalse();
assertThat(firstRetryDestinationTopic.isReusableRetryTopic()).isFalse();
assertThat(firstRetryDestinationTopic.getDestinationDelay()).isEqualTo(1000);
assertThat(firstRetryDestinationTopic.getDestinationPartitions()).isEqualTo(numPartitions);
@@ -169,7 +164,6 @@ class DestinationTopicPropertiesFactoryTests {
assertThat(secondRetryProperties.isDltTopic()).isFalse();
assertThat(secondRetryProperties.isRetryTopic()).isTrue();
DestinationTopic secondRetryDestinationTopic = destinationTopicList.get(2);
assertThat(secondRetryDestinationTopic.isSingleTopicRetry()).isFalse();
assertThat(secondRetryDestinationTopic.isReusableRetryTopic()).isFalse();
assertThat(secondRetryDestinationTopic.getDestinationDelay()).isEqualTo(2000);
assertThat(secondRetryDestinationTopic.getDestinationPartitions()).isEqualTo(numPartitions);
@@ -193,7 +187,7 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues, classifier,
numPartitions, kafkaOperations, fixedDelayStrategy, noDltStrategy,
numPartitions, kafkaOperations, noDltStrategy,
TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE, multipleTopicsSameIntervalReuseStrategy,
RetryTopicConstants.NOT_SET).createProperties();
@@ -202,50 +196,6 @@ class DestinationTopicPropertiesFactoryTests {
assertThat(propertiesList.get(2).isDltTopic()).isFalse();
}
@Test
@SuppressWarnings("deprecation")
void shouldCreateOneRetryPropertyForFixedBackoffWithSingleTopicStrategy() {
// when
FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(1000);
int maxAttempts = 5;
List<Long> backOffValues = new BackOffValuesGenerator(maxAttempts, backOffPolicy).generateValues();
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations, FixedDelayStrategy.SINGLE_TOPIC,
dltStrategy, suffixWithDelayValueSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy,
-1).createProperties();
List<DestinationTopic> destinationTopicList = propertiesList
.stream()
.map(properties -> new DestinationTopic("mainTopic" + properties.suffix(), properties))
.collect(Collectors.toList());
// then
assertThat(propertiesList.size() == 3).isTrue();
DestinationTopic mainDestinationTopic = destinationTopicList.get(0);
assertThat(mainDestinationTopic.isMainTopic()).isTrue();
DestinationTopic.Properties firstRetryProperties = propertiesList.get(1);
assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix);
assertThat(firstRetryProperties.isRetryTopic()).isTrue();
DestinationTopic retryDestinationTopic = destinationTopicList.get(1);
assertThat(retryDestinationTopic.isSingleTopicRetry()).isTrue();
assertThat(retryDestinationTopic.isReusableRetryTopic()).isFalse();
assertThat(retryDestinationTopic.getDestinationDelay()).isEqualTo(1000);
DestinationTopic.Properties dltProperties = propertiesList.get(2);
assertThat(dltProperties.suffix()).isEqualTo(dltSuffix);
assertThat(dltProperties.isDltTopic()).isTrue();
DestinationTopic dltTopic = destinationTopicList.get(2);
assertThat(dltTopic.getDestinationDelay()).isEqualTo(0);
assertThat(dltTopic.getDestinationPartitions()).isEqualTo(numPartitions);
}
@Test
@SuppressWarnings("deprecation")
void shouldCreateOneRetryPropertyForFixedBackoffWithSingleTopicSameIntervalReuseStrategy() {
@@ -259,7 +209,7 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations, FixedDelayStrategy.MULTIPLE_TOPICS,
classifier, numPartitions, kafkaOperations,
dltStrategy, suffixWithDelayValueSuffixingStrategy, singleTopicSameIntervalReuseStrategy,
-1).createProperties();
@@ -277,8 +227,7 @@ class DestinationTopicPropertiesFactoryTests {
DestinationTopic.Properties firstRetryProperties = propertiesList.get(1);
assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix);
DestinationTopic retryDestinationTopic = destinationTopicList.get(1);
assertThat(retryDestinationTopic.isSingleTopicRetry()).isTrue();
assertThat(retryDestinationTopic.isReusableRetryTopic()).isFalse();
assertThat(retryDestinationTopic.isReusableRetryTopic()).isTrue();
assertThat(retryDestinationTopic.getDestinationDelay()).isEqualTo(1000);
DestinationTopic.Properties dltProperties = propertiesList.get(2);
@@ -303,7 +252,6 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.MULTIPLE_TOPICS,
dltStrategy, suffixWithDelayValueSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy,
-1).createProperties();
@@ -322,14 +270,12 @@ class DestinationTopicPropertiesFactoryTests {
assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix + "-0");
assertThat(firstRetryProperties.isRetryTopic()).isTrue();
DestinationTopic retryDestinationTopic = destinationTopicList.get(1);
assertThat(retryDestinationTopic.isSingleTopicRetry()).isFalse();
assertThat(retryDestinationTopic.isReusableRetryTopic()).isFalse();
assertThat(retryDestinationTopic.getDestinationDelay()).isEqualTo(5000);
DestinationTopic.Properties secondRetryProperties = propertiesList.get(2);
assertThat(secondRetryProperties.suffix()).isEqualTo(retryTopicSuffix + "-1");
DestinationTopic secondRetryDestinationTopic = destinationTopicList.get(2);
assertThat(secondRetryDestinationTopic.isSingleTopicRetry()).isFalse();
assertThat(secondRetryDestinationTopic.isReusableRetryTopic()).isFalse();
assertThat(secondRetryDestinationTopic.getDestinationDelay()).isEqualTo(5000);
@@ -354,7 +300,6 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.SINGLE_TOPIC,
dltStrategy, suffixWithIndexTopicSuffixingStrategy,
multipleTopicsSameIntervalReuseStrategy, -1).createProperties();
@@ -377,7 +322,6 @@ class DestinationTopicPropertiesFactoryTests {
List<DestinationTopic.Properties> propertiesList =
new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.MULTIPLE_TOPICS,
dltStrategy, suffixWithIndexTopicSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy,
-1).createProperties();
@@ -402,7 +346,6 @@ class DestinationTopicPropertiesFactoryTests {
// when
DestinationTopicPropertiesFactory factory = new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix,
backOffValues, classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.MULTIPLE_TOPICS,
dltStrategy, suffixWithDelayValueSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy, -1);
List<DestinationTopic.Properties> propertiesList = factory.createProperties();
@@ -433,7 +376,6 @@ class DestinationTopicPropertiesFactoryTests {
// when
DestinationTopicPropertiesFactory factory = new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix,
backOffValues, classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.MULTIPLE_TOPICS,
dltStrategy, suffixWithDelayValueSuffixingStrategy, singleTopicSameIntervalReuseStrategy, -1);
List<DestinationTopic.Properties> propertiesList = factory.createProperties();
@@ -463,7 +405,6 @@ class DestinationTopicPropertiesFactoryTests {
// when
DestinationTopicPropertiesFactory factory = new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix,
backOffValues, classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.MULTIPLE_TOPICS,
dltStrategy, suffixWithIndexTopicSuffixingStrategy, singleTopicSameIntervalReuseStrategy, -1);
List<DestinationTopic.Properties> propertiesList = factory.createProperties();
@@ -489,7 +430,6 @@ class DestinationTopicPropertiesFactoryTests {
// when
DestinationTopicPropertiesFactory factory = new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix,
backOffValues, classifier, numPartitions, kafkaOperations,
FixedDelayStrategy.SINGLE_TOPIC,
dltStrategy, suffixWithDelayValueSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy, -1);
List<DestinationTopic.Properties> propertiesList = factory.createProperties();
@@ -514,7 +454,6 @@ class DestinationTopicPropertiesFactoryTests {
assertThat(topicProperties.isRetryTopic()).isTrue();
DestinationTopic topic = new DestinationTopic("irrelevant" + topicProperties.suffix(), topicProperties);
assertThat(topic.isDltTopic()).isFalse();
assertThat(topic.isSingleTopicRetry()).isEqualTo(expectedIsSingleTopicRetry);
assertThat(topic.isReusableRetryTopic()).isEqualTo(expectedReusableTopic);
assertThat(topic.getDestinationDelay()).isEqualTo(expectedDelay);
assertThat(topic.getDestinationPartitions()).isEqualTo(numPartitions);
@@ -522,4 +461,5 @@ class DestinationTopicPropertiesFactoryTests {
assertThat(topic.shouldRetryOn(maxAttempts, new IllegalArgumentException())).isFalse();
assertThat(topic.shouldRetryOn(0, new RuntimeException())).isFalse();
}
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicTests.java
View File

@@ -119,7 +119,7 @@ public class DestinationTopicTests {
@SuppressWarnings("deprecation")
protected DestinationTopic.Properties singleFixedRetryTopicProps4 =
new DestinationTopic.Properties(1000, retrySuffix, DestinationTopic.Type.SINGLE_TOPIC_RETRY, 4, 1,
new DestinationTopic.Properties(1000, retrySuffix, DestinationTopic.Type.REUSABLE_RETRY_TOPIC, 4, 1,
DltStrategy.ALWAYS_RETRY_ON_ERROR, kafkaOperations2, getShouldRetryOn(), timeout);
protected DestinationTopic.Properties dltTopicProps4 =

5
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicExceptionRoutingIntegrationTests.java
View File

@@ -248,7 +248,8 @@ public class RetryTopicExceptionRoutingIntegrationTests {
CountDownLatchContainer container;
@SuppressWarnings("deprecation")
@RetryableTopic(fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC, backoff = @Backoff(50))
@RetryableTopic(sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC,
backoff = @Backoff(50))
@KafkaListener(topics = FRAMEWORK_FATAL_EXCEPTION_TOPIC)
public void listenWithAnnotation(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
container.fatalFrameworkLatch.countDown();
@@ -337,7 +338,7 @@ public class RetryTopicExceptionRoutingIntegrationTests {
.newInstance()
.fixedBackOff(50)
.includeTopic(ONLY_RETRY_VIA_TOPIC)
.useSingleTopicForFixedDelays()
.sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
.doNotRetryOnDltFailure()
.dltHandlerMethod("dltProcessorWithError", DLT_METHOD_NAME)
.create(template);

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java
View File

@@ -548,7 +548,7 @@ public class RetryTopicIntegrationTests {
.fixedBackOff(50)
.maxAttempts(5)
.concurrency(1)
.useSingleTopicForFixedDelays()
.useSingleTopicForSameIntervals()
.includeTopic(FIRST_TOPIC)
.doNotRetryOnDltFailure()
.dltHandlerMethod("myCustomDltProcessor", DLT_METHOD_NAME)

7
spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -142,14 +142,15 @@ class EnableKafkaKotlinTests {
}
val eh = object: CommonErrorHandler {
override fun handleRecord(
override fun handleOne(
thrownException: Exception,
record: ConsumerRecord<*, *>,
consumer: Consumer<*, *>,
container: MessageListenerContainer
) {
): Boolean {
error = true
latch2.countDown()
return true
}
override fun handleBatch(