From e2daa2da8cc745d41a98bd5a2b4f577bf79b0d99 Mon Sep 17 00:00:00 2001 From: "Kim, Joo Hyuk" Date: Tue, 26 Mar 2024 00:11:06 +0900 Subject: [PATCH] Add `checkstyle` rule on `empty-line-separater` * Update checkstyle.xml * Fix checkstyle violations * Fix test class checkstyle * Add field also. --- .../kafka/test/EmbeddedKafkaKraftBroker.java | 1 - .../kafka/test/assertj/KafkaConditions.java | 3 +-- .../test/condition/EmbeddedKafkaCondition.java | 3 +-- .../EmbeddedKafkaContextCustomizerTests.java | 5 +---- ...fkaListenerAnnotationBeanPostProcessor.java | 1 - .../RetryTopicConfigurationProvider.java | 1 - .../config/KafkaListenerEndpointRegistry.java | 3 --- .../MultiMethodKafkaListenerEndpoint.java | 1 - .../core/DefaultKafkaConsumerFactory.java | 1 - .../core/DefaultKafkaProducerFactory.java | 3 --- .../kafka/core/KafkaTemplate.java | 2 -- .../kafka/core/MicrometerProducerListener.java | 3 +-- .../kafka/event/KafkaEvent.java | 3 +-- .../kafka/listener/BatchMessageListener.java | 3 +-- .../CommonContainerStoppingErrorHandler.java | 3 +-- .../listener/CommonLoggingErrorHandler.java | 3 +-- .../ConsumerAwareRebalanceListener.java | 3 +-- .../listener/ConsumerAwareRecordRecoverer.java | 3 +-- .../DeadLetterPublishingRecoverer.java | 3 ++- .../KafkaMessageListenerContainer.java | 2 -- .../BatchMessagingMessageListenerAdapter.java | 3 +-- .../adapter/DelegatingInvocableHandler.java | 1 - .../RecordMessagingMessageListenerAdapter.java | 3 +-- .../DefaultDestinationTopicResolver.java | 1 - .../kafka/retrytopic/DestinationTopic.java | 1 + .../retrytopic/RetryTopicConfiguration.java | 2 ++ .../RetryTopicConfigurationBuilder.java | 2 -- .../kafka/retrytopic/RetryTopicConfigurer.java | 1 - .../KafkaStreamsMicrometerListener.java | 3 +-- .../support/AbstractKafkaHeaderMapper.java | 3 +-- .../support/DefaultKafkaHeaderMapper.java | 1 - .../kafka/support/KafkaUtils.java | 1 - .../kafka/support/Suffixer.java | 3 +-- .../BatchMessagingMessageConverter.java | 3 +-- .../converter/MessagingMessageConverter.java | 3 +-- .../micrometer/KafkaListenerObservation.java | 3 +-- .../serializer/DelegatingSerializer.java | 3 +-- .../support/serializer/JsonDeserializer.java | 3 +-- .../support/serializer/SerializationUtils.java | 1 - .../support/serializer/ToStringSerializer.java | 3 +-- .../EnableKafkaIntegrationTests.java | 3 ++- .../KafkaListenerEndpointRegistryTests.java | 1 - .../core/DefaultKafkaConsumerFactoryTests.java | 1 - .../core/DefaultKafkaProducerFactoryTests.java | 1 - ...efaultTransactionIdSuffixStrategyTests.java | 1 - .../kafka/core/KafkaTemplateTests.java | 5 ++--- ...rrentMessageListenerContainerMockTests.java | 1 - ...oncurrentMessageListenerContainerTests.java | 1 - .../ConsumerAwareRebalanceListenerTests.java | 4 +--- .../kafka/listener/ConsumerSeekAwareTests.java | 3 ++- .../listener/ContainerGroupSequencerTests.java | 3 +-- .../ErrorHandlingDeserializerTests.java | 3 ++- .../kafka/listener/ErrorHandlingUtilsTest.java | 11 ++++++++++- .../KafkaMessageListenerContainerTests.java | 5 ++++- .../listener/ManualNackPauseResumeTests.java | 3 ++- .../kafka/listener/ManualNackRecordTests.java | 3 ++- .../ManualNackRecordZeroSleepTests.java | 3 ++- ...tainerWhileErrorHandlerIsRetryingTests.java | 3 +-- .../listener/TransactionalContainerTests.java | 1 - .../MessagingMessageListenerAdapterTests.java | 1 + .../retrytopic/DestinationTopicTests.java | 4 +++- ...ngRetryTopicClassLevelIntegrationTests.java | 4 ++++ .../ExistingRetryTopicIntegrationTests.java | 6 +++++- ...istenerContainerFactoryConfigurerTests.java | 1 - .../ListenerContainerFactoryResolverTests.java | 4 +++- ...sLevelExceptionRoutingIntegrationTests.java | 18 +++++++++++++++++- .../RetryTopicClassLevelIntegrationTests.java | 18 +++++++++++++++++- .../retrytopic/RetryTopicConfigurerTests.java | 3 ++- ...yTopicExceptionRoutingIntegrationTests.java | 17 +++++++++++++++++ .../retrytopic/RetryTopicIntegrationTests.java | 17 +++++++++++++++++ ...icSameContainerFactoryIntegrationTests.java | 9 +++++++++ ...fkaStreamsInteractiveQueryServiceTests.java | 1 - .../ObservationIntegrationTests.java | 3 +-- .../serializer/JsonSerializationTests.java | 4 ++-- src/checkstyle/checkstyle.xml | 6 ++++++ 75 files changed, 161 insertions(+), 105 deletions(-) diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java index c414aa2e..e5635fbd 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java @@ -204,7 +204,6 @@ public class EmbeddedKafkaKraftBroker implements EmbeddedKafkaBroker { } } - private void start() { if (this.cluster != null) { return; diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java index cb1bbf1f..b995f204 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2024 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. @@ -90,7 +90,6 @@ public final class KafkaConditions { return new ConsumerRecordPartitionCondition(partition); } - public static class ConsumerRecordKeyCondition extends Condition> { private final K key; diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java index 103f0366..a5b779b0 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -211,7 +211,6 @@ public class EmbeddedKafkaCondition implements ExecutionCondition, AfterAllCallb return parent.getStore(Namespace.create(getClass(), parent)); } - public static EmbeddedKafkaBroker getBroker() { return BROKERS.get(); } diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java index e79cdaa9..b3bc00c3 100644 --- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -48,7 +48,6 @@ public class EmbeddedKafkaContextCustomizerTests { AnnotationUtils.findAnnotation(SecondTestWithEmbeddedKafka.class, EmbeddedKafka.class); } - @Test void testHashCode() { assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode()).isNotEqualTo(0); @@ -56,7 +55,6 @@ public class EmbeddedKafkaContextCustomizerTests { .isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass).hashCode()); } - @Test void testEquals() { assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass)) @@ -93,7 +91,6 @@ public class EmbeddedKafkaContextCustomizerTests { .matches("127.0.0.1:[0-9]+,127.0.0.1:[0-9]+"); } - @EmbeddedKafka(kraft = false) private static final class TestWithEmbeddedKafka { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java index 00fe8041..42c3bfbb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java @@ -1190,7 +1190,6 @@ public class KafkaListenerAnnotationBeanPostProcessor private record BytesToStringConverter(Charset charset) implements Converter { - @Override public String convert(byte[] source) { return new String(source, this.charset); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryTopicConfigurationProvider.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryTopicConfigurationProvider.java index 85cdc2f4..1b85aacb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryTopicConfigurationProvider.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryTopicConfigurationProvider.java @@ -37,7 +37,6 @@ import org.springframework.core.log.LogAccessor; import org.springframework.kafka.retrytopic.RetryTopicConfiguration; import org.springframework.lang.Nullable; - /** * * Attempts to provide an instance of diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java index 6f1c8f0f..daf046fa 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java @@ -361,7 +361,6 @@ public class KafkaListenerEndpointRegistry implements ListenerContainerRegistry, return listenerContainer; } - @Override public void destroy() { for (MessageListenerContainer listenerContainer : getListenerContainers()) { @@ -419,7 +418,6 @@ public class KafkaListenerEndpointRegistry implements ListenerContainerRegistry, return this.running; } - @Override public void onApplicationEvent(ContextRefreshedEvent event) { if (event.getApplicationContext().equals(this.applicationContext)) { @@ -439,7 +437,6 @@ public class KafkaListenerEndpointRegistry implements ListenerContainerRegistry, } } - private static final class AggregatingCallback implements Runnable { private final AtomicInteger count; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java index f06ea580..9cb90d44 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java @@ -61,7 +61,6 @@ public class MultiMethodKafkaListenerEndpoint extends MethodKafkaListenerE setBean(bean); } - /** * Get a method list. * @return the method list. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java index f027627f..c44e1055 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java @@ -252,7 +252,6 @@ public class DefaultKafkaConsumerFactory extends KafkaResourceFactory this.valueDeserializerSupplier = valueDeserializerSupplier(valueDeserializerSupplier); } - /** * Set to false (default true) to prevent programmatically provided deserializers (via * constructor or setters) from being configured using the producer configuration, diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java index 5a9792cd..17a91bc9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java @@ -68,8 +68,6 @@ import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.StringUtils; - - /** * The {@link ProducerFactory} implementation for a {@code singleton} shared {@link Producer} instance. *

@@ -567,7 +565,6 @@ public class DefaultKafkaProducerFactory extends KafkaResourceFactory return newFactory; } - /** * Ensures that the returned properties map contains a transaction id prefix. * The {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index f6377ed9..0b16ead2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -308,7 +308,6 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo ((MessagingMessageConverter) this.messageConverter).setMessagingConverter(messageConverter); } - @Override public boolean isTransactional() { return this.transactional; @@ -590,7 +589,6 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo return observeSend((ProducerRecord) producerRecord); } - @Override public List partitionsFor(String topic) { Producer producer = getTheProducer(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/MicrometerProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/core/MicrometerProducerListener.java index 9a458457..34af1905 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/MicrometerProducerListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/MicrometerProducerListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2022 the original author or authors. + * Copyright 2020-2024 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. @@ -65,7 +65,6 @@ public class MicrometerProducerListener implements ProducerFactory.Listene this.tags = tags; } - @Override public synchronized void producerAdded(String id, Producer producer) { if (!this.metrics.containsKey(id)) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java index 144cf3ef..6f359b93 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2023 the original author or authors. + * Copyright 2015-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,7 +19,6 @@ package org.springframework.kafka.event; import org.springframework.context.ApplicationEvent; import org.springframework.util.Assert; - /** * Base class for events. * diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchMessageListener.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchMessageListener.java index 19f5cac2..bc71bbb4 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchMessageListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2024 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. @@ -40,7 +40,6 @@ import org.springframework.lang.Nullable; @FunctionalInterface public interface BatchMessageListener extends GenericMessageListener>> { - /** * Listener receives the original {@link ConsumerRecords} object instead of a * list of {@link ConsumerRecord}. 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 4b783e7c..374bbc3e 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-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -83,7 +83,6 @@ public class CommonContainerStoppingErrorHandler extends KafkaExceptionLogLevelA stopContainer(container, thrownException); } - @Override public void handleRemaining(Exception thrownException, List> records, Consumer consumer, MessageListenerContainer container) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonLoggingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonLoggingErrorHandler.java index 93d5174d..9e731c05 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonLoggingErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonLoggingErrorHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -47,7 +47,6 @@ public class CommonLoggingErrorHandler implements CommonErrorHandler { this.ackAfterHandle = ackAfterHandle; } - @Override public boolean handleOne(Exception thrownException, ConsumerRecord record, Consumer consumer, MessageListenerContainer container) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java index 4a1b1fff..12a12b10 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -34,7 +34,6 @@ import org.apache.kafka.common.TopicPartition; */ public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener { - /** * The same as {@link #onPartitionsRevoked(Collection)} with the additional consumer * parameter. It is invoked by the container before any pending offsets are committed. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordRecoverer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordRecoverer.java index e44cdd5b..bc9fae7c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordRecoverer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordRecoverer.java @@ -1,5 +1,5 @@ /* - * Copyright 2021 the original author or authors. + * Copyright 2021-2024 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. @@ -32,7 +32,6 @@ import org.springframework.lang.Nullable; @FunctionalInterface public interface ConsumerAwareRecordRecoverer extends ConsumerRecordRecoverer { - @Override default void accept(ConsumerRecord record, Exception exception) { accept(record, null, exception); 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 54400edb..7ff214c0 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2023 the original author or authors. + * Copyright 2018-2024 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. @@ -1364,6 +1364,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement this.exceptionCauseFqcn = exceptionCauseFqcn; return this; } + /** * Sets the name of the header that will be used to store the keyExceptionMessage * of the original record. 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 0bca9827..336f13ee 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 @@ -137,7 +137,6 @@ import org.springframework.util.StringUtils; import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationRegistry; - /** * Single-threaded Message listener container using the Java {@link Consumer} supporting * auto-partition assignment or user-configured assignment. @@ -593,7 +592,6 @@ public class KafkaMessageListenerContainer // NOSONAR line count + "]"; } - private final class ListenerConsumer implements SchedulingAwareRunnable, ConsumerSeekCallback { private static final String COMMITTING = "Committing: "; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java index b7241c0f..e9ad1ccb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2024 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. @@ -35,7 +35,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; - /** * A {@link org.springframework.kafka.listener.MessageListener MessageListener} * adapter that invokes a configurable {@link HandlerAdapter}; used when the factory is diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java index 9f94c33d..3770dc6e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java @@ -47,7 +47,6 @@ import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.util.Assert; import org.springframework.validation.Validator; - /** * Delegates to an {@link InvocableHandlerMethod} based on the message payload type. * Matches a single, non-annotated parameter or one that is annotated with diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java index ee9b7910..6caa854e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2024 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. @@ -28,7 +28,6 @@ import org.springframework.kafka.support.converter.ProjectingMessageConverter; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; - /** * A {@link org.springframework.kafka.listener.MessageListener MessageListener} * adapter that invokes a configurable {@link HandlerAdapter}; used when the factory is 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 1b4d8931..3f8fc01c 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 @@ -39,7 +39,6 @@ import org.springframework.kafka.retrytopic.DestinationTopic.Type; import org.springframework.lang.Nullable; import org.springframework.util.Assert; - /** * * Default implementation of the {@link DestinationTopicResolver} interface. 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 936a1c4d..3725b2ff 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 @@ -148,6 +148,7 @@ public class DestinationTopic { @Nullable private final Boolean autoStartDltHandler; + /** * Create an instance with the provided properties with the DLT container starting * automatically (if the container factory is so configured). diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java index 9a163144..cdb5c89d 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java @@ -91,7 +91,9 @@ public class RetryTopicConfiguration { static class TopicCreation { private final boolean shouldCreateTopics; + private final int numPartitions; + private final short replicationFactor; TopicCreation(@Nullable Boolean shouldCreate, @Nullable Integer numPartitions, @Nullable Short replicationFactor) { 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 b0c3cfd4..d52c1c1e 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 @@ -72,7 +72,6 @@ public class RetryTopicConfigurationBuilder { private RetryTopicConfiguration.TopicCreation topicCreationConfiguration = new RetryTopicConfiguration.TopicCreation(); - private ConcurrentKafkaListenerContainerFactory listenerContainerFactory; @Nullable @@ -401,7 +400,6 @@ public class RetryTopicConfigurationBuilder { return this; } - /** * Configure a {@link FixedBackOffPolicy}. * @param interval the interval. 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 e1233788..433e7f20 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 @@ -43,7 +43,6 @@ import org.springframework.kafka.support.TopicForRetryable; import org.springframework.lang.NonNull; import org.springframework.lang.Nullable; - /** * *

Configures main, retry and DLT topics based on a main endpoint and provided diff --git a/spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsMicrometerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsMicrometerListener.java index e1814c13..d290e3e2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsMicrometerListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsMicrometerListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2022 the original author or authors. + * Copyright 2020-2024 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. @@ -64,7 +64,6 @@ public class KafkaStreamsMicrometerListener implements StreamsBuilderFactoryBean this.tags = tags; } - @Override public synchronized void streamsAdded(String id, KafkaStreams kafkaStreams) { if (!this.metrics.containsKey(id)) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java index f33b135f..0a1af826 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2024 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. @@ -271,7 +271,6 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper { return null; } - /** * A matcher for headers. * @since 2.3 diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java index 66b81ac9..25bf914b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java @@ -439,7 +439,6 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper { this.untrustedType = untrustedType; } - public void setHeaderValue(byte[] headerValue) { // NOSONAR this.headerValue = headerValue; // NOSONAR array reference } 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 29438b98..b5894451 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 @@ -63,7 +63,6 @@ public final class KafkaUtils { */ public static final String VALUE_DESERIALIZER_EXCEPTION_HEADER = DESERIALIZER_EXCEPTION_HEADER_PREFIX + "Value"; - private static Function, String> prFormatter = ProducerRecord::toString; private static Function, String> crFormatter = diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/Suffixer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/Suffixer.java index e2cbd543..85f5bce0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/Suffixer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/Suffixer.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2024 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. @@ -22,7 +22,6 @@ import java.util.stream.Collectors; import org.springframework.util.Assert; import org.springframework.util.StringUtils; - /** * Utility class that suffixes strings. * diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java index cc644b0d..7afea777 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2022 the original author or authors. + * Copyright 2016-2024 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. @@ -238,7 +238,6 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter { return record.value() == null ? KafkaNull.INSTANCE : record.value(); } - /** * Convert the record value. * @param record the record. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java index 51a7148e..339bd17c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2024 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. @@ -136,7 +136,6 @@ public class MessagingMessageConverter implements RecordMessageConverter { this.rawRecordHeader = rawRecordHeader; } - protected org.springframework.messaging.converter.MessageConverter getMessagingConverter() { return this.messagingConverter; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java index e3055fa7..1bbcfbc5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-2024 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. @@ -36,7 +36,6 @@ public enum KafkaListenerObservation implements ObservationDocumentation { */ LISTENER_OBSERVATION { - @Override public Class> getDefaultConvention() { return DefaultKafkaListenerObservationConvention.class; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java index b7e9c377..0e2e0de0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2022 the original author or authors. + * Copyright 2019-2024 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. @@ -187,7 +187,6 @@ public class DelegatingSerializer implements Serializer { throw new UnsupportedOperationException(); } - @Override public byte[] serialize(String topic, Headers headers, Object data) { if (data == null) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java index 14b4a218..e72e08bf 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2023 the original author or authors. + * Copyright 2015-2024 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. @@ -178,7 +178,6 @@ public class JsonDeserializer implements Deserializer { this(targetType, true); } - /** * Construct an instance with the provided target type, and a default {@link ObjectMapper}. * @param targetType the target java type to use if no type info headers are present. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java index 3c2259af..12bc17c5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java @@ -241,7 +241,6 @@ public final class SerializationUtils { return super.resolveClass(desc); } - }; return (DeserializationException) ois.readObject(); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java index b2691c08..f80e33cc 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2024 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. @@ -43,7 +43,6 @@ public class ToStringSerializer implements Serializer { */ public static final String ADD_TYPE_INFO_HEADERS = "spring.message.add.type.headers"; - /** * Header for the type of key. */ 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 d1c9191c..55dd270f 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 @@ -1440,6 +1440,7 @@ public class EnableKafkaIntegrationTests { public SeekToOffsetFromComputeFunction seekToOffsetFromComputeFunction() { return new SeekToOffsetFromComputeFunction(); } + @Bean public IfaceListener ifaceListener() { return new IfaceListenerImpl(); @@ -2351,6 +2352,7 @@ public class EnableKafkaIntegrationTests { public static class SeekToOffsetFromComputeFunction extends AbstractConsumerSeekAware { CountDownLatch latch1 = new CountDownLatch(10); + CountDownLatch latch2 = new CountDownLatch(1); @KafkaListener(id = "seekToComputeFn", topics = "seekToComputeFn") @@ -2651,7 +2653,6 @@ public class EnableKafkaIntegrationTests { private String bar; - public Foo() { } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/config/KafkaListenerEndpointRegistryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/config/KafkaListenerEndpointRegistryTests.java index 3522a26e..2576f38b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/config/KafkaListenerEndpointRegistryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/config/KafkaListenerEndpointRegistryTests.java @@ -71,7 +71,6 @@ public class KafkaListenerEndpointRegistryTests { .withMessage("'idMatcher' cannot be null"); } - @DisplayName("getListenerContainersMatching with BiPredicate throws on null biPredicate") @Test void getListenerContainersMatchingBiPredicateThrowsOnNullBiPredicate() { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java index 30002846..fbf9e796 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java @@ -119,7 +119,6 @@ public class DefaultKafkaConsumerFactoryTests { DefaultKafkaConsumerFactory target = new DefaultKafkaConsumerFactory(originalConfig) { - @Override protected Consumer createRawConsumer(Map configProps) { configPassedToKafkaConsumer.putAll(configProps); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java index 383e0e96..cfd2d327 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java @@ -706,7 +706,6 @@ public class DefaultKafkaProducerFactoryTests { assertThat(configPassedToKafkaConsumer.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG)).isEqualTo("tx.1"); } - @SuppressWarnings({ "rawtypes", "unchecked" }) @Test void configUpdates() { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultTransactionIdSuffixStrategyTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultTransactionIdSuffixStrategyTests.java index 1f8db2a1..20256d3d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultTransactionIdSuffixStrategyTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultTransactionIdSuffixStrategyTests.java @@ -50,7 +50,6 @@ class DefaultTransactionIdSuffixStrategyTests { assertThat(suffix).isNotNull(); } - @Test void acquireSuffixWithCacheExhausted() { String txIdPrefix = "txIdPrefix"; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index 98ab9c3c..4c311547 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2024 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. @@ -132,7 +132,6 @@ public class KafkaTemplateTests { private static final ProducerPostProcessor noopProducerPostProcessor = processor -> processor; - @BeforeAll public static void setUp() { embeddedKafka = EmbeddedKafkaCondition.getBroker(); @@ -349,6 +348,7 @@ public class KafkaTemplateTests { } } + PL pl1 = new PL(); PL pl2 = new PL(); CompositeProducerListener cpl = new CompositeProducerListener<>(new PL[]{ pl1, pl2 }); @@ -638,5 +638,4 @@ public class KafkaTemplateTests { .withMessage("Offset supplied in TopicPartitionOffset is invalid: " + tpoWithNullOffset); } - } 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 7b573bf2..0a59a00b 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 @@ -1413,7 +1413,6 @@ public class ConcurrentMessageListenerContainerMockTests { this.latch.countDown(); } - @Override public void onIdleContainer(Map assignments, ConsumerSeekCallback callback) { if (latch.getCount() > 0) { 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 4db13f47..3a8ba407 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 @@ -652,7 +652,6 @@ public class ConcurrentMessageListenerContainerTests { } - @Test public void testAckOnErrorRecord() throws Exception { logger.info("Start ack on error"); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListenerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListenerTests.java index 605f5b33..5930ea8e 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListenerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListenerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -46,7 +46,6 @@ public class ConsumerAwareRebalanceListenerTests { assertThat(called.get()).isTrue(); } - @Test void nonConsumerAwareTestRevoked() { AtomicBoolean called = new AtomicBoolean(); @@ -61,7 +60,6 @@ public class ConsumerAwareRebalanceListenerTests { assertThat(called.get()).isTrue(); } - @Test void nonConsumerAwareTestLost() { AtomicBoolean called = new AtomicBoolean(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerSeekAwareTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerSeekAwareTests.java index 14425789..d579423f 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerSeekAwareTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerSeekAwareTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2023 the original author or authors. + * Copyright 2020-2024 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. @@ -46,6 +46,7 @@ public class ConsumerSeekAwareTests { void beginningEndAndBulkSeekToTimestamp() throws Exception { class CSA extends AbstractConsumerSeekAware { } + AbstractConsumerSeekAware csa = new CSA(); var exec1 = Executors.newSingleThreadExecutor(); var exec2 = Executors.newSingleThreadExecutor(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerGroupSequencerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerGroupSequencerTests.java index fd08de11..344f7938 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerGroupSequencerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerGroupSequencerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021 the original author or authors. + * Copyright 2021-2024 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. @@ -100,7 +100,6 @@ public class ContainerGroupSequencerTests { final List receivedAt = Collections.synchronizedList(new ArrayList<>()); - @KafkaListener(id = "one", topics = "ContainerGroupSequencerTests", containerGroup = "g1", concurrency = "2") public void listen1(String in) { LOGGER.debug(in); 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 876c76c9..e06b309b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2023 the original author or authors. + * Copyright 2018-2024 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. @@ -130,6 +130,7 @@ public class ErrorHandlingDeserializerTests { } } + ErrorHandlingDeserializer ehd = new ErrorHandlingDeserializer<>(new MyDes()); Headers headers = new RecordHeaders(); ehd.deserialize("foo", headers, new byte[1]); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingUtilsTest.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingUtilsTest.java index c0e576b1..0be4a13d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingUtilsTest.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingUtilsTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2023 the original author or authors. + * Copyright 2017-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -53,15 +53,24 @@ import org.springframework.util.backoff.FixedBackOff; class ErrorHandlingUtilsTest { private final Exception thrownException = new RuntimeException("initial cause"); + private final Consumer consumer = mock(Consumer.class); + private final MessageListenerContainer container = mock(MessageListenerContainer.class); + private final Runnable listener = mock(Runnable.class); + private final BackOff backOff = new FixedBackOff(1000, 3); + private final CommonErrorHandler seeker = mock(CommonErrorHandler.class); + @SuppressWarnings("unchecked") private final BiConsumer, Exception> recoverer = mock(BiConsumer.class); + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(ErrorHandlingUtilsTest.class)); + private final List retryListeners = new ArrayList<>(); + private final BinaryExceptionClassifier classifier = BinaryExceptionClassifier.defaultClassifier(); private final ConsumerRecords consumerRecords = recordsOf( 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 1f243bc7..8eb81d94 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 @@ -1448,7 +1448,6 @@ public class KafkaMessageListenerContainerTests { throw new IllegalStateException(); } - }); final CountDownLatch commitLatch = new CountDownLatch(1); @@ -1520,6 +1519,7 @@ public class KafkaMessageListenerContainerTests { } } + Listener messageListener = new Listener(); containerProps.setMessageListener(messageListener); containerProps.setSyncCommits(true); @@ -1599,6 +1599,7 @@ public class KafkaMessageListenerContainerTests { } } + Listener messageListener = new Listener(); containerProps.setMessageListener(messageListener); containerProps.setSyncCommits(true); @@ -2607,6 +2608,7 @@ public class KafkaMessageListenerContainerTests { } } + Listener messageListener = new Listener(); containerProps.setMessageListener(messageListener); containerProps.setMissingTopicsFatal(false); @@ -2935,6 +2937,7 @@ public class KafkaMessageListenerContainerTests { } } + containerProps.setMessageListener(new Listener()); containerProps.setMissingTopicsFatal(false); KafkaMessageListenerContainer container = diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackPauseResumeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackPauseResumeTests.java index 4df3e491..7875e575 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackPauseResumeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackPauseResumeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2022 the original author or authors. + * Copyright 2017-2024 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. @@ -77,6 +77,7 @@ public class ManualNackPauseResumeTests { @SuppressWarnings("rawtypes") @Autowired private Consumer consumer; + @Autowired private Config config; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java index a5427e6d..b6e3024e 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2022 the original author or authors. + * Copyright 2017-2024 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. @@ -77,6 +77,7 @@ public class ManualNackRecordTests { @SuppressWarnings("rawtypes") @Autowired private Consumer consumer; + @Autowired private Config config; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordZeroSleepTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordZeroSleepTests.java index 586c5989..3cb52fbe 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordZeroSleepTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordZeroSleepTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-2024 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. @@ -76,6 +76,7 @@ public class ManualNackRecordZeroSleepTests { @SuppressWarnings("rawtypes") @Autowired private Consumer consumer; + @Autowired private Config config; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/PauseContainerWhileErrorHandlerIsRetryingTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/PauseContainerWhileErrorHandlerIsRetryingTests.java index 3d4a3bc9..7e920052 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/PauseContainerWhileErrorHandlerIsRetryingTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/PauseContainerWhileErrorHandlerIsRetryingTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-2024 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. @@ -153,7 +153,6 @@ public class PauseContainerWhileErrorHandlerIsRetryingTests { } } - @KafkaListener(id = "id", groupId = "grp", topics = "foo") public void process(List batch, Acknowledgment acknowledgment) { batch.forEach((msg) -> { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index e204fdef..9982ddf8 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java @@ -1086,7 +1086,6 @@ public class TransactionalContainerTests { assertThatIllegalStateException().isThrownBy(container::start); } - @Test void testArbpWithoutRecovery() throws InterruptedException { // init producer diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapterTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapterTests.java index 800c399c..a1c59bc5 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapterTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapterTests.java @@ -62,6 +62,7 @@ public class MessagingMessageListenerAdapterTests { } } + MyAdapter adapter = new MyAdapter(); adapter.setFallbackType(String.class); RecordMessageConverter converter = mock(RecordMessageConverter.class); 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 d9078690..e3da1a8b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2023 the original author or authors. + * Copyright 2018-2024 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. @@ -303,7 +303,9 @@ public class DestinationTopicTests { } class PropsHolder { + final String topicName; + final DestinationTopic.Properties props; PropsHolder(String topicName, DestinationTopic.Properties props) { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicClassLevelIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicClassLevelIntegrationTests.java index 20c944cc..d937518e 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicClassLevelIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicClassLevelIntegrationTests.java @@ -203,13 +203,17 @@ class ExistingRetryTopicClassLevelIntegrationTests { static class CountDownLatchContainer { CountDownLatch countDownLatch1 = new CountDownLatch(12); + CountDownLatch countDownLatch2 = new CountDownLatch(12); + } static class CountByPartitionContainer { Map mainTopicMessageCountByPartition = new HashMap<>(); + Map retryTopicMessageCountByPartition = new HashMap<>(); + } @Configuration diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicIntegrationTests.java index 327ab60f..6241ef77 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -200,13 +200,17 @@ public class ExistingRetryTopicIntegrationTests { static class CountDownLatchContainer { CountDownLatch countDownLatch1 = new CountDownLatch(40); + CountDownLatch countDownLatch2 = new CountDownLatch(40); + } static class CountByPartitionContainer { Map mainTopicMessageCountByPartition = new HashMap<>(); + Map retryTopicMessageCountByPartition = new HashMap<>(); + } @Configuration diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java index 672d0b64..b5749175 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java @@ -139,7 +139,6 @@ class ListenerContainerFactoryConfigurerTests { @Mock private KafkaListenerEndpoint endpoint; - @Test void shouldSetupErrorHandling() { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryResolverTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryResolverTests.java index 8179ac2d..b3f8129d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryResolverTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryResolverTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2024 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. @@ -56,7 +56,9 @@ class ListenerContainerFactoryResolverTests { private ConcurrentKafkaListenerContainerFactory factoryFromDefaultBeanName; private final static String factoryName = "testListenerContainerFactory"; + private final static String otherFactoryName = "otherTestListenerContainerFactory"; + private final static String defaultFactoryBeanName = "defaultTestListenerContainerFactory"; @Test diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelExceptionRoutingIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelExceptionRoutingIntegrationTests.java index ca89fbe4..6e0eb0ea 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelExceptionRoutingIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelExceptionRoutingIntegrationTests.java @@ -75,9 +75,13 @@ import org.springframework.util.backoff.FixedBackOff; class RetryTopicClassLevelExceptionRoutingIntegrationTests { final static String BLOCKING_AND_TOPIC_RETRY = "blocking-and-topic-retry"; + final static String ONLY_RETRY_VIA_BLOCKING = "only-retry-blocking-topic"; + final static String ONLY_RETRY_VIA_TOPIC = "only-retry-topic"; + final static String USER_FATAL_EXCEPTION_TOPIC = "user-fatal-topic"; + final static String FRAMEWORK_FATAL_EXCEPTION_TOPIC = "framework-fatal-topic"; @Autowired @@ -238,7 +242,6 @@ class RetryTopicClassLevelExceptionRoutingIntegrationTests { @Autowired CountDownLatchContainer container; - @KafkaHandler public void listenWithAnnotation(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { container.fatalFrameworkLatch.countDown(); @@ -257,20 +260,33 @@ class RetryTopicClassLevelExceptionRoutingIntegrationTests { static class CountDownLatchContainer { CountDownLatch blockingAndTopicsLatch = new CountDownLatch(12); + CountDownLatch onlyRetryViaBlockingLatch = new CountDownLatch(4); + CountDownLatch onlyRetryViaTopicLatch = new CountDownLatch(3); + CountDownLatch fatalUserLatch = new CountDownLatch(1); + CountDownLatch fatalFrameworkLatch = new CountDownLatch(1); + CountDownLatch annotatedDltOnlyBlockingLatch = new CountDownLatch(1); + CountDownLatch annotatedDltUserFatalLatch = new CountDownLatch(1); + CountDownLatch annotatedDltFrameworkFatalLatch = new CountDownLatch(1); + CountDownLatch dltProcessorLatch = new CountDownLatch(1); + CountDownLatch dltProcessorWithErrorLatch = new CountDownLatch(1); AtomicInteger blockingAndTopicsListenerInvocations = new AtomicInteger(); + AtomicInteger onlyRetryViaTopicListenerInvocations = new AtomicInteger(); + AtomicInteger onlyRetryViaBlockingListenerInvocations = new AtomicInteger(); + AtomicInteger userFatalListenerInvocations = new AtomicInteger(); + AtomicInteger fatalFrameworkListenerInvocations = new AtomicInteger(); } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelIntegrationTests.java index bf99d248..53390934 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicClassLevelIntegrationTests.java @@ -480,7 +480,6 @@ class RetryTopicClassLevelIntegrationTests { @Autowired CountDownLatchContainer container; - @KafkaHandler public void listenWithAnnotation2(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { container.countDownIfNotKnown(receivedTopic, container.countDownLatchNoRetry); @@ -557,22 +556,39 @@ class RetryTopicClassLevelIntegrationTests { static class CountDownLatchContainer { CountDownLatch countDownLatch1 = new CountDownLatch(5); + CountDownLatch countDownLatch2 = new CountDownLatch(3); + CountDownLatch countDownLatch3 = new CountDownLatch(3); + CountDownLatch countDownLatch4 = new CountDownLatch(4); + CountDownLatch countDownLatch51 = new CountDownLatch(4); + CountDownLatch countDownLatch52 = new CountDownLatch(4); + CountDownLatch countDownLatch6 = new CountDownLatch(4); + CountDownLatch countDownLatchNoRetry = new CountDownLatch(1); + CountDownLatch countDownLatchDltOne = new CountDownLatch(1); + CountDownLatch countDownLatchDltTwo = new CountDownLatch(1); + CountDownLatch countDownLatchDltThree = new CountDownLatch(1); + CountDownLatch countDownLatchDltFour = new CountDownLatch(1); + CountDownLatch countDownLatchReuseOne = new CountDownLatch(2); + CountDownLatch countDownLatchReuseTwo = new CountDownLatch(5); + CountDownLatch countDownLatchReuseThree = new CountDownLatch(5); + CountDownLatch customDltCountdownLatch = new CountDownLatch(1); + CountDownLatch customErrorHandlerCountdownLatch = new CountDownLatch(6); + CountDownLatch customMessageConverterCountdownLatch = new CountDownLatch(6); final List knownTopics = new ArrayList<>(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java index 82c98264..37c70df2 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java @@ -357,6 +357,7 @@ class RetryTopicConfigurerTests { } static class NoOpsClass { - void noOpsMethod() { }; + void noOpsMethod() { } } + } 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 8366936b..934e5841 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 @@ -77,9 +77,13 @@ public class RetryTopicExceptionRoutingIntegrationTests { private static final Logger logger = LoggerFactory.getLogger(RetryTopicExceptionRoutingIntegrationTests.class); public final static String BLOCKING_AND_TOPIC_RETRY = "blocking-and-topic-retry"; + public final static String ONLY_RETRY_VIA_BLOCKING = "only-retry-blocking-topic"; + public final static String ONLY_RETRY_VIA_TOPIC = "only-retry-topic"; + public final static String USER_FATAL_EXCEPTION_TOPIC = "user-fatal-topic"; + public final static String FRAMEWORK_FATAL_EXCEPTION_TOPIC = "framework-fatal-topic"; @Autowired @@ -270,20 +274,33 @@ public class RetryTopicExceptionRoutingIntegrationTests { static class CountDownLatchContainer { CountDownLatch blockingAndTopicsLatch = new CountDownLatch(12); + CountDownLatch onlyRetryViaBlockingLatch = new CountDownLatch(4); + CountDownLatch onlyRetryViaTopicLatch = new CountDownLatch(3); + CountDownLatch fatalUserLatch = new CountDownLatch(1); + CountDownLatch fatalFrameworkLatch = new CountDownLatch(1); + CountDownLatch annotatedDltOnlyBlockingLatch = new CountDownLatch(1); + CountDownLatch annotatedDltUserFatalLatch = new CountDownLatch(1); + CountDownLatch annotatedDltFrameworkFatalLatch = new CountDownLatch(1); + CountDownLatch dltProcessorLatch = new CountDownLatch(1); + CountDownLatch dltProcessorWithErrorLatch = new CountDownLatch(1); AtomicInteger blockingAndTopicsListenerInvocations = new AtomicInteger(); + AtomicInteger onlyRetryViaTopicListenerInvocations = new AtomicInteger(); + AtomicInteger onlyRetryViaBlockingListenerInvocations = new AtomicInteger(); + AtomicInteger userFatalListenerInvocations = new AtomicInteger(); + AtomicInteger fatalFrameworkListenerInvocations = new AtomicInteger(); } 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 15ca2bdc..5dd6de3b 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 @@ -571,22 +571,39 @@ public class RetryTopicIntegrationTests { static class CountDownLatchContainer { CountDownLatch countDownLatch1 = new CountDownLatch(5); + CountDownLatch countDownLatch2 = new CountDownLatch(3); + CountDownLatch countDownLatch3 = new CountDownLatch(3); + CountDownLatch countDownLatch4 = new CountDownLatch(4); + CountDownLatch countDownLatch51 = new CountDownLatch(4); + CountDownLatch countDownLatch52 = new CountDownLatch(4); + CountDownLatch countDownLatch6 = new CountDownLatch(4); + CountDownLatch countDownLatchNoRetry = new CountDownLatch(1); + CountDownLatch countDownLatchDltOne = new CountDownLatch(1); + CountDownLatch countDownLatchDltTwo = new CountDownLatch(1); + CountDownLatch countDownLatchDltThree = new CountDownLatch(1); + CountDownLatch countDownLatchDltFour = new CountDownLatch(1); + CountDownLatch countDownLatchReuseOne = new CountDownLatch(2); + CountDownLatch countDownLatchReuseTwo = new CountDownLatch(5); + CountDownLatch countDownLatchReuseThree = new CountDownLatch(5); + CountDownLatch customDltCountdownLatch = new CountDownLatch(1); + CountDownLatch customErrorHandlerCountdownLatch = new CountDownLatch(6); + CountDownLatch customMessageConverterCountdownLatch = new CountDownLatch(6); final List knownTopics = new ArrayList<>(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicSameContainerFactoryIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicSameContainerFactoryIntegrationTests.java index 836a8036..3917de3b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicSameContainerFactoryIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicSameContainerFactoryIntegrationTests.java @@ -246,20 +246,29 @@ public class RetryTopicSameContainerFactoryIntegrationTests { static class CountDownLatchContainer { CountDownLatch countDownLatchFirstRetryable = new CountDownLatch(4); + CountDownLatch countDownLatchSecondRetryable = new CountDownLatch(3); + CountDownLatch countDownLatchDltOne = new CountDownLatch(1); + CountDownLatch countDownLatchDltSecond = new CountDownLatch(1); CountDownLatch countDownLatchBasic = new CountDownLatch(1); + CountDownLatch customizerLatch = new CountDownLatch(10); CountDownLatch countDownLatchClassLevelFirstRetryable = new CountDownLatch(4); + CountDownLatch countDownLatchClassLevelSecondRetryable = new CountDownLatch(3); + CountDownLatch countDownLatchClassLevelDltOne = new CountDownLatch(1); + CountDownLatch countDownLatchClassLevelDltSecond = new CountDownLatch(1); CountDownLatch countDownLatchClassLevelBasic = new CountDownLatch(1); + CountDownLatch customizerClassLevelLatch = new CountDownLatch(10); + } @EnableKafka diff --git a/spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java b/spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java index 5865133e..3cbf9d5b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java @@ -200,7 +200,6 @@ class KafkaStreamsInteractiveQueryServiceTests { serializer); } - @Configuration @EnableKafka @EnableKafkaStreams diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationIntegrationTests.java index ce3935b3..f6bb6335 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2023 the original author or authors. + * Copyright 2022-2024 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. @@ -115,7 +115,6 @@ public class ObservationIntegrationTests extends SampleTestRunner { }; } - @Configuration @EnableKafka public static class Config { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerializationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerializationTests.java index ff04ce17..a2280488 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerializationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerializationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2024 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. @@ -516,7 +516,7 @@ public class JsonSerializationTests { Child(int number) { super(number); } + } - } diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index 313da7c2..d746bba5 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -191,6 +191,12 @@ + + + + + +