From ea643a90f4a03b4c4338b0382ff7362682939f90 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 6 Sep 2023 10:26:28 -0700 Subject: [PATCH] Remove deprecated interfaces, classes, and methods --- .../src/main/asciidoc/changes-since-1.0.adoc | 109 +++++++ .../src/main/asciidoc/whats-new.adoc | 107 +------ .../kafka/annotation/RetryableTopic.java | 8 - .../RetryableTopicAnnotationProcessor.java | 1 - ...AbstractKafkaListenerContainerFactory.java | 67 +---- .../core/DefaultKafkaProducerFactory.java | 28 -- .../kafka/core/KafkaOperations2.java | 281 ------------------ .../kafka/core/ProducerFactory.java | 23 -- .../AbstractMessageListenerContainer.java | 54 ---- .../kafka/listener/BatchErrorHandler.java | 64 ---- .../CommonContainerStoppingErrorHandler.java | 8 +- .../CommonDelegatingErrorHandler.java | 11 +- .../kafka/listener/CommonErrorHandler.java | 72 ++--- .../listener/CommonMixedErrorHandler.java | 8 +- .../ConcurrentMessageListenerContainer.java | 1 - .../ConsumerAwareBatchErrorHandler.java | 52 ---- .../listener/ConsumerAwareErrorHandler.java | 53 ---- .../ConsumerAwareRecordInterceptor.java | 36 --- .../ContainerAwareBatchErrorHandler.java | 62 ---- .../listener/ContainerAwareErrorHandler.java | 52 ---- .../DeadLetterPublishingRecoverer.java | 2 +- .../kafka/listener/DefaultErrorHandler.java | 8 +- .../kafka/listener/ErrorHandler.java | 46 --- .../kafka/listener/ErrorHandlerAdapter.java | 176 ----------- .../kafka/listener/ErrorHandlingUtils.java | 30 -- .../kafka/listener/GenericErrorHandler.java | 82 ----- .../KafkaMessageListenerContainer.java | 48 +-- .../ListenerInvokingBatchErrorHandler.java | 47 --- .../kafka/listener/ListenerUtils.java | 69 +---- .../RemainingRecordsErrorHandler.java | 61 ---- .../DefaultDestinationTopicResolver.java | 9 +- .../kafka/retrytopic/DestinationTopic.java | 23 +- .../DestinationTopicPropertiesFactory.java | 53 ++-- .../kafka/retrytopic/FixedDelayStrategy.java | 41 --- .../RetryTopicConfigurationBuilder.java | 31 +- .../retrytopic/RetryTopicConfigurer.java | 20 -- .../kafka/streams/HeaderEnricher.java | 115 ------- .../messaging/MessagingTransformer.java | 106 ------- .../kafka/support/KafkaUtils.java | 2 +- .../support/micrometer/MicrometerHolder.java | 17 -- .../EnableKafkaIntegrationTests.java | 3 +- ...rentMessageListenerContainerMockTests.java | 2 +- ...ncurrentMessageListenerContainerTests.java | 5 +- .../ErrorHandlingDeserializerTests.java | 3 +- .../KafkaMessageListenerContainerTests.java | 43 +-- .../RemainingRecordsErrorHandlerTests.java | 4 +- ...estinationTopicPropertiesFactoryTests.java | 72 +---- .../retrytopic/DestinationTopicTests.java | 2 +- ...TopicExceptionRoutingIntegrationTests.java | 5 +- .../RetryTopicIntegrationTests.java | 2 +- .../kafka/listener/EnableKafkaKotlinTests.kt | 7 +- 51 files changed, 196 insertions(+), 2035 deletions(-) delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations2.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordInterceptor.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerInvokingBatchErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/retrytopic/FixedDelayStrategy.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/streams/HeaderEnricher.java delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/streams/messaging/MessagingTransformer.java diff --git a/spring-kafka-docs/src/main/asciidoc/changes-since-1.0.adoc b/spring-kafka-docs/src/main/asciidoc/changes-since-1.0.adoc index c80869b8..725720a3 100644 --- a/spring-kafka-docs/src/main/asciidoc/changes-since-1.0.adoc +++ b/spring-kafka-docs/src/main/asciidoc/changes-since-1.0.adoc @@ -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 <> 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 <> for more information. + +[[x30-Native]] +==== Native Images + +Support for creating native images is provided. +See <> 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 <> 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 <> 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 <> 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 <> 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` parameter. + +[[x30-lc-changes]] +==== Listener Container Changes + +Events related to consumer authentication and authorization failures are now published by the container. +See <> for more information. + +You can now customize the thread names used by consumer threads. +See <> for more information. + +The container property `restartAfterAuthException` has been added. +See <> for more information. + +[[x30-template-changes]] +==== `KafkaTemplate` Changes + +The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s. +See <>. + +[[x30-rkt-changes]] +==== `ReplyingKafkaTemplate` Changes + +The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s. +See <> and <>. + +[[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 <> for more information. + +You can now manually commit parts of a batch before the entire batch is processed. +See <> 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 <> 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]] diff --git a/spring-kafka-docs/src/main/asciidoc/whats-new.adoc b/spring-kafka-docs/src/main/asciidoc/whats-new.adoc index 3aa9ac46..03090874 100644 --- a/spring-kafka-docs/src/main/asciidoc/whats-new.adoc +++ b/spring-kafka-docs/src/main/asciidoc/whats-new.adoc @@ -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 <>. [[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 <> 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 <> for more information. - -[[x30-Native]] -==== Native Images - -Support for creating native images is provided. -See <> 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 <> 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 <> 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 <> 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 <> 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` parameter. - -[[x30-lc-changes]] -==== Listener Container Changes - -Events related to consumer authentication and authorization failures are now published by the container. -See <> for more information. - -You can now customize the thread names used by consumer threads. -See <> for more information. - -The container property `restartAfterAuthException` has been added. -See <> for more information. - -[[x30-template-changes]] -==== `KafkaTemplate` Changes - -The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s. -See <>. - -[[x30-rkt-changes]] -==== `ReplyingKafkaTemplate` Changes - -The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s. -See <> and <>. - -[[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 <> for more information. - -You can now manually commit parts of a batch before the entire batch is processed. -See <> 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 <> 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). diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java index e0137c25..168cff43 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java @@ -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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java index d10032c4..673e62b9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java @@ -143,7 +143,6 @@ public class RetryableTopicAnnotationProcessor { .retryOn(includes) .notRetryOn(excludes) .traversingCauses(traverse) - .useSingleTopicForFixedDelays(annotation.fixedDelayTopicStrategy()) .dltProcessingFailureStrategy(annotation.dltStrategy()) .autoStartDltHandler(autoStartDlt) .setTopicSuffixingStrategy(annotation.topicSuffixingStrategy()) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index 20c70ba3..faf83afb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -26,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, K, V> - implements KafkaListenerContainerFactory, ApplicationEventPublisherAware, InitializingBean, - ApplicationContextAware { + implements KafkaListenerContainerFactory, 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 consumerFactory; @@ -155,23 +149,6 @@ public abstract class AbstractKafkaListenerContainerFactory "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 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 getKeySerializer() { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations2.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations2.java deleted file mode 100644 index f22ce235..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations2.java +++ /dev/null @@ -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 the key type. - * @param 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 { - - /** - * 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> 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> 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> 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> 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> 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> 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> 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> 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> send(ProducerRecord 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> send(Message message); - - /** - * See {@link Producer#partitionsFor(String)}. - * @param topic the topic. - * @return the partition info. - */ - List partitionsFor(String topic); - - /** - * See {@link Producer#metrics()}. - * @return the metrics. - */ - Map metrics(); - - /** - * Execute some arbitrary operation(s) on the producer and return the result. - * @param callback the callback. - * @param the result type. - * @return the result. - */ - @Nullable - T execute(ProducerCallback 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 the result type. - * @return the result. - */ - @Nullable - T executeInTransaction(OperationsCallback 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 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 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 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 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 receive(Collection 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 receive(Collection requested, Duration pollTimeout); - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java index bb3dcdc7..5da1d21a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java @@ -78,29 +78,6 @@ public interface ProducerFactory { 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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index 83a1ed1a..1ad004c3 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -72,8 +72,6 @@ public abstract class AbstractMessageListenerContainer implements GenericMessageListenerContainer, 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 private ApplicationEventPublisher applicationEventPublisher; - @SuppressWarnings("deprecation") - private GenericErrorHandler errorHandler; - private CommonErrorHandler commonErrorHandler; private boolean autoStartup = true; @@ -227,55 +222,6 @@ public abstract class AbstractMessageListenerContainer return this.applicationEventPublisher; } - /** - * Set the error handler to call when the listener throws an exception. - * @param errorHandler the error handler. - * @since 2.2 - * @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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java deleted file mode 100644 index 46ebfa83..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java +++ /dev/null @@ -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> { - - /** - * 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); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonContainerStoppingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonContainerStoppingErrorHandler.java index 516aa74e..4b783e7c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonContainerStoppingErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonContainerStoppingErrorHandler.java @@ -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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java index adb78a17..9d0deac2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java @@ -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(), diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java index 0174eb16..752ca309 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java @@ -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> 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())); } /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java index 5c81fb00..a5b604c3 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java @@ -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(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index c7d34ab9..904ddaa5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -272,7 +272,6 @@ public class ConcurrentMessageListenerContainer 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()); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java deleted file mode 100644 index 00e321e7..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java +++ /dev/null @@ -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); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareErrorHandler.java deleted file mode 100644 index 8f1ddcf3..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareErrorHandler.java +++ /dev/null @@ -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> data, Consumer consumer, - MessageListenerContainer container) { - handle(thrownException, null, consumer); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordInterceptor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordInterceptor.java deleted file mode 100644 index 2fb5a9ae..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordInterceptor.java +++ /dev/null @@ -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 the key type. - * @param 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 extends RecordInterceptor { - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java deleted file mode 100644 index a6abb18c..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java +++ /dev/null @@ -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); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java deleted file mode 100644 index ec1021ab..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java +++ /dev/null @@ -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> records, - Consumer consumer) { - - throw new UnsupportedOperationException("Container should never call this"); - } - - @Override - void handle(Exception thrownException, @Nullable List> records, Consumer consumer, - MessageListenerContainer container); - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java index 99a4ac2c..2eb2ed20 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java @@ -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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java index 1b5a9ad6..8d3df390 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java @@ -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(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java deleted file mode 100644 index 3463a717..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java +++ /dev/null @@ -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> { - - /** - * 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> records, Consumer consumer, - MessageListenerContainer container) { - handle(thrownException, null); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java deleted file mode 100644 index 68feb09a..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java +++ /dev/null @@ -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> 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 partitions, - Runnable publishPause) { - - if (this.batchErrorHandler instanceof FallbackBatchErrorHandler) { - ((FallbackBatchErrorHandler) this.batchErrorHandler).onPartitionsAssigned(consumer, partitions, - publishPause); - } - } - -} - diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java index 88614ee4..a154d629 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java @@ -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, Exception> recoverer, LogAccessor logger, - KafkaException.Level logLevel, List 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 diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java deleted file mode 100644 index 85e3613c..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java +++ /dev/null @@ -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 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 { - - /** - * 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"); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 5c7fd3d8..350de4df 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -1012,40 +1012,13 @@ public class KafkaMessageListenerContainer // 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 // 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 // NOSONAR line count ConsumerRecords 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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerInvokingBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerInvokingBatchErrorHandler.java deleted file mode 100644 index 45013b22..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerInvokingBatchErrorHandler.java +++ /dev/null @@ -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); - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java index 173a9337..b398966d 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java @@ -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 executions, ThreadLocal lastIntervals, MessageListenerContainer container) throws InterruptedException { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java deleted file mode 100644 index a262e92e..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java +++ /dev/null @@ -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> records, Consumer consumer); - - @Override - default void handle(Exception thrownException, @Nullable List> records, - Consumer consumer, MessageListenerContainer container) { - - handle(thrownException, records, consumer); - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java index 5786bc65..522256b4 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java @@ -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(); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java index 18ffa3f6..c6895be5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java @@ -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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java index 24763da0..9a1356af 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java @@ -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 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 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 shouldRetryOn) { + BiPredicate 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, diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/FixedDelayStrategy.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/FixedDelayStrategy.java deleted file mode 100644 index e256ad1a..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/FixedDelayStrategy.java +++ /dev/null @@ -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 - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java index bdf6b809..cd431b65 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java @@ -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 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(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java index ccc057e7..78814748 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java @@ -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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/streams/HeaderEnricher.java b/spring-kafka/src/main/java/org/springframework/kafka/streams/HeaderEnricher.java deleted file mode 100644 index 5a9854e3..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/streams/HeaderEnricher.java +++ /dev/null @@ -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 the key type. - * @param the value type. - * - * @author Gary Russell - * @since 2.3 - * @deprecated in favor of {@link HeaderEnricherProcessor}. - * - */ -@Deprecated -public class HeaderEnricher implements Transformer> { - - private final Map headerExpressions = new HashMap<>(); - - private ProcessorContext processorContext; - - public HeaderEnricher(Map headerExpressions) { - this.headerExpressions.putAll(headerExpressions); - } - - @Override - public void init(ProcessorContext context) { - this.processorContext = context; - } - - @Override - public KeyValue transform(K key, V value) { - Headers headers = this.processorContext.headers(); - Container 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 the key type. - * @param the value type. - * - */ - public static final class Container { - - 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; - } - - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/streams/messaging/MessagingTransformer.java b/spring-kafka/src/main/java/org/springframework/kafka/streams/messaging/MessagingTransformer.java deleted file mode 100644 index 102bf0c6..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/streams/messaging/MessagingTransformer.java +++ /dev/null @@ -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 the key type. - * @param the value type. - * @param the result value type. - * - * @author Gary Russell - * @since 2.3 - * @deprecated in favor of {@link MessagingProcessor}. - * - */ -@Deprecated -public class MessagingTransformer implements Transformer> { - - 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 transform(K key, V value) { - Headers headers = this.processorContext.headers(); - ConsumerRecord record = new ConsumerRecord(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 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 - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java index 08f8a998..79dd2e83 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java @@ -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) { } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java index 9e73a7c5..78274592 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java @@ -53,23 +53,6 @@ public final class MicrometerHolder { private final Function> 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 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. diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 1dc31aa7..daaccb72 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -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")); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java index f48f36ec..59657c0b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java @@ -156,7 +156,7 @@ public class ConcurrentMessageListenerContainerMockTests { container.setCommonErrorHandler(new CommonErrorHandler() { @Override - public boolean remainingRecords() { + public boolean seeksAfterHandling() { return true; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index f216c5ea..bfff9e77 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -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(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java index 1dff27db..ea6d8e60 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java @@ -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; } }); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index e7c62539..abb812a1 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -1479,16 +1479,17 @@ public class KafkaMessageListenerContainerTests { containerProps.setClientId("clientId"); KafkaMessageListenerContainer 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 props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); - ContainerProperties containerProps = new ContainerProperties("foo"); - containerProps.setMissingTopicsFatal(false); - KafkaMessageListenerContainer badContainer = - new KafkaMessageListenerContainer<>(cf, containerProps); - badContainer.setBatchErrorHandler((thrownException, data) -> { - }); - badContainer.setupMessageListener((MessageListener) m -> { - }); - assertThatIllegalStateException().isThrownBy(() -> badContainer.start()) - .withMessageContaining("Error handler is not compatible with the message listener"); - - } - - @Test - @SuppressWarnings("deprecation") - public void testBadBatchErrorHandler() { - Map props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); - ContainerProperties containerProps = new ContainerProperties("foo"); - containerProps.setMissingTopicsFatal(false); - KafkaMessageListenerContainer badContainer = - new KafkaMessageListenerContainer<>(cf, containerProps); - badContainer.setErrorHandler((thrownException, data) -> { - }); - badContainer.setupMessageListener((BatchMessageListener) 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"); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java index 53990684..a4d7e96f 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java @@ -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; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactoryTests.java index 2e08aaeb..e83bcec1 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactoryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactoryTests.java @@ -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 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 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 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 backOffValues = new BackOffValuesGenerator(maxAttempts, backOffPolicy).generateValues(); - - List propertiesList = - new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues, - classifier, numPartitions, kafkaOperations, FixedDelayStrategy.SINGLE_TOPIC, - dltStrategy, suffixWithDelayValueSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy, - -1).createProperties(); - - List 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 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 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 propertiesList = new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues, classifier, numPartitions, kafkaOperations, - FixedDelayStrategy.SINGLE_TOPIC, dltStrategy, suffixWithIndexTopicSuffixingStrategy, multipleTopicsSameIntervalReuseStrategy, -1).createProperties(); @@ -377,7 +322,6 @@ class DestinationTopicPropertiesFactoryTests { List 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 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 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 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 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(); } + } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicTests.java index 8cad3c7a..1b381df4 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicTests.java @@ -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 = diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicExceptionRoutingIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicExceptionRoutingIntegrationTests.java index 425dfbd2..d8b71572 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicExceptionRoutingIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicExceptionRoutingIntegrationTests.java @@ -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); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java index 44dcb091..537bf23d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java @@ -548,7 +548,7 @@ public class RetryTopicIntegrationTests { .fixedBackOff(50) .maxAttempts(5) .concurrency(1) - .useSingleTopicForFixedDelays() + .useSingleTopicForSameIntervals() .includeTopic(FIRST_TOPIC) .doNotRetryOnDltFailure() .dltHandlerMethod("myCustomDltProcessor", DLT_METHOD_NAME) diff --git a/spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt b/spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt index ef1a4728..9a112cb8 100644 --- a/spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt +++ b/spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt @@ -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(