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

Migrate test classes to AssertJ (#1724)

Browse Source 
* Modify @RetryableListener Tests to use AssertJ

* Migrate to AssertJ for all test cases.
This commit is contained in:
Aravind Ranganathan
2021-03-05 11:38:10 -05:00
committed by GitHub
parent 5dffa83981
commit 1cafb9f381
19 changed files with 350 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaConsumerBackoffManagerTest.java
View File

@@ -16,9 +16,8 @@
package org.springframework.kafka.listener;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowableOfType;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.times;
@@ -79,14 +78,14 @@ class KafkaConsumerBackoffManagerTest {
backoffManager.createContext(dueTimestamp, testListenerId, topicPartition);
// given
KafkaBackoffException backoffException = assertThrows(KafkaBackoffException.class,
() -> backoffManager.maybeBackoff(context));
KafkaBackoffException backoffException = catchThrowableOfType(() -> backoffManager.maybeBackoff(context),
KafkaBackoffException.class);
// then
assertEquals(dueTimestamp, backoffException.getDueTimestamp());
assertEquals(testListenerId, backoffException.getListenerId());
assertEquals(topicPartition, backoffException.getTopicPartition());
assertEquals(context, backoffManager.getBackoff(topicPartition));
assertThat(backoffException.getDueTimestamp()).isEqualTo(dueTimestamp);
assertThat(backoffException.getListenerId()).isEqualTo(testListenerId);
assertThat(backoffException.getTopicPartition()).isEqualTo(topicPartition);
assertThat(backoffManager.getBackoff(topicPartition)).isEqualTo(context);
then(listenerContainer).should(times(1)).pausePartition(topicPartition);
}
@@ -102,7 +101,7 @@ class KafkaConsumerBackoffManagerTest {
backoffManager.maybeBackoff(context);
// then
assertNull(backoffManager.getBackoff(topicPartition));
assertThat(backoffManager.getBackoff(topicPartition)).isNull();
then(listenerContainer).should(times(0)).pausePartition(topicPartition);
}
@@ -121,7 +120,7 @@ class KafkaConsumerBackoffManagerTest {
backoffManager.onApplicationEvent(partitionIdleEvent);
// then
assertEquals(context, backoffManager.getBackoff(topicPartition));
assertThat(backoffManager.getBackoff(topicPartition)).isEqualTo(context);
then(listenerContainer).should(times(0)).resumePartition(topicPartition);
}
@@ -140,7 +139,7 @@ class KafkaConsumerBackoffManagerTest {
backoffManager.onApplicationEvent(partitionIdleEvent);
// then
assertNull(backoffManager.getBackoff(topicPartition));
assertThat(backoffManager.getBackoff(topicPartition)).isNull();
then(listenerContainer).should(times(1)).resumePartition(topicPartition);
}
}

10
spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapterTest.java
View File

@@ -16,7 +16,7 @@
package org.springframework.kafka.listener.adapter;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
@@ -140,7 +140,7 @@ class KafkaBackoffAwareMessageListenerAdapterTest {
// then
then(kafkaConsumerBackoffManager).should(times(1))
.createContext(timestampCaptor.capture(), eq(listenerId), eq(topicPartition));
assertEquals(originalTimestamp, timestampCaptor.getValue());
assertThat(timestampCaptor.getValue()).isEqualTo(originalTimestamp);
then(kafkaConsumerBackoffManager).should(times(1))
.maybeBackoff(context);
@@ -162,7 +162,7 @@ class KafkaBackoffAwareMessageListenerAdapterTest {
// then
then(kafkaConsumerBackoffManager).should(times(1))
.createContext(timestampCaptor.capture(), eq(listenerId), eq(topicPartition));
assertEquals(originalTimestamp, timestampCaptor.getValue());
assertThat(timestampCaptor.getValue()).isEqualTo(originalTimestamp);
then(kafkaConsumerBackoffManager).should(times(1))
.maybeBackoff(context);
@@ -184,7 +184,7 @@ class KafkaBackoffAwareMessageListenerAdapterTest {
// then
then(kafkaConsumerBackoffManager).should(times(1))
.createContext(timestampCaptor.capture(), eq(listenerId), eq(topicPartition));
assertEquals(originalTimestamp, timestampCaptor.getValue());
assertThat(timestampCaptor.getValue()).isEqualTo(originalTimestamp);
then(kafkaConsumerBackoffManager).should(times(1))
.maybeBackoff(context);
@@ -207,7 +207,7 @@ class KafkaBackoffAwareMessageListenerAdapterTest {
// then
then(kafkaConsumerBackoffManager).should(times(1))
.createContext(timestampCaptor.capture(), eq(listenerId), eq(topicPartition));
assertEquals(originalTimestamp, timestampCaptor.getValue());
assertThat(timestampCaptor.getValue()).isEqualTo(originalTimestamp);
then(kafkaConsumerBackoffManager).should(times(1))
.maybeBackoff(context);

10
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/BackOffValuesGeneratorTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.List;
@@ -45,8 +45,8 @@ class BackOffValuesGeneratorTests {
List<Long> backOffValues = backOffValuesGenerator.generateValues();
// then
List<Long> expectedBackoffs = Arrays.asList(1000L, 1000L);
assertEquals(expectedBackoffs, backOffValues);
List<Long> expectedBackOffs = Arrays.asList(1000L, 1000L);
assertThat(backOffValues).isEqualTo(expectedBackOffs);
}
@Test
@@ -63,7 +63,7 @@ class BackOffValuesGeneratorTests {
// then
List<Long> expectedBackoffs = Arrays.asList(1000L, 2000L, 4000L);
assertEquals(expectedBackoffs, backOffValues);
assertThat(backOffValues).isEqualTo(expectedBackoffs);
}
@Test
@@ -78,6 +78,6 @@ class BackOffValuesGeneratorTests {
// then
List<Long> expectedBackoffs = Arrays.asList(0L, 0L, 0L);
assertEquals(expectedBackoffs, backOffValues);
assertThat(backOffValues).isEqualTo(expectedBackoffs);
}
}

36
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactoryTests.java
View File

@@ -16,9 +16,8 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
@@ -121,18 +120,18 @@ class DeadLetterPublishingRecovererFactoryTests {
// then
then(kafkaOperations2).should(times(1)).send(producerRecordCaptor.capture());
ProducerRecord producerRecord = producerRecordCaptor.getValue();
assertEquals(testRetryTopic, producerRecord.topic());
assertEquals(value, producerRecord.value());
assertEquals(key, producerRecord.key());
assertEquals(2, producerRecord.partition());
assertThat(producerRecord.topic()).isEqualTo(testRetryTopic);
assertThat(producerRecord.value()).isEqualTo(value);
assertThat(producerRecord.key()).isEqualTo(key);
assertThat(producerRecord.partition()).isEqualTo(2);
// assert headers
Header attemptsHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
assertNotNull(attemptsHeader);
assertEquals(2, attemptsHeader.value()[0]);
assertThat(attemptsHeader).isNotNull();
assertThat(attemptsHeader.value()[0]).isEqualTo(Integer.valueOf(2).byteValue());
Header timestampHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);
assertNotNull(timestampHeader);
assertEquals(this.nowTimestamp, new BigInteger(timestampHeader.value()).longValue());
assertThat(timestampHeader).isNotNull();
assertThat(new BigInteger(timestampHeader.value()).longValue()).isEqualTo(this.nowTimestamp);
}
@Test
@@ -164,8 +163,8 @@ class DeadLetterPublishingRecovererFactoryTests {
then(kafkaOperations2).should(times(1)).send(producerRecordCaptor.capture());
ProducerRecord producerRecord = producerRecordCaptor.getValue();
Header attemptsHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
assertNotNull(attemptsHeader);
assertEquals(2, attemptsHeader.value()[0]);
assertThat(attemptsHeader).isNotNull();
assertThat(attemptsHeader.value()[0]).isEqualTo(Integer.valueOf(2).byteValue());
}
@Test
@@ -197,8 +196,8 @@ class DeadLetterPublishingRecovererFactoryTests {
then(kafkaOperations2).should(times(1)).send(producerRecordCaptor.capture());
ProducerRecord producerRecord = producerRecordCaptor.getValue();
Header originalTimestampHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP);
assertNotNull(originalTimestampHeader);
assertEquals(this.nowTimestamp, new BigInteger(originalTimestampHeader.value()).longValue());
assertThat(originalTimestampHeader).isNotNull();
assertThat(new BigInteger(originalTimestampHeader.value()).longValue()).isEqualTo(this.nowTimestamp);
}
@Test
@@ -231,8 +230,8 @@ class DeadLetterPublishingRecovererFactoryTests {
then(kafkaOperations2).should(times(1)).send(producerRecordCaptor.capture());
ProducerRecord producerRecord = producerRecordCaptor.getValue();
Header originalTimestampHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP);
assertNotNull(originalTimestampHeader);
assertEquals(timestamp, new BigInteger(originalTimestampHeader.value()).longValue());
assertThat(originalTimestampHeader).isNotNull();
assertThat(new BigInteger(originalTimestampHeader.value()).longValue()).isEqualTo(timestamp);
}
@Test
@@ -261,7 +260,8 @@ class DeadLetterPublishingRecovererFactoryTests {
// when
DeadLetterPublishingRecoverer deadLetterPublishingRecoverer = factory.create();
assertThrows(NestedRuntimeException.class, () -> deadLetterPublishingRecoverer.accept(this.consumerRecord, e));
assertThatExceptionOfType(NestedRuntimeException.class)
.isThrownBy(() -> deadLetterPublishingRecoverer.accept(this.consumerRecord, e));
// then
then(kafkaOperations2).should(times(0)).send(any(ProducerRecord.class));

109
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicProcessorTests.java
View File

@@ -16,8 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.then;
import java.util.ArrayList;
@@ -58,7 +57,7 @@ class DefaultDestinationTopicProcessorTests extends DestinationTopicTests {
destinationTopicProcessor.processDestinationTopicProperties(props -> processedProps.add(props), context);
// then
assertEquals(allProps, processedProps);
assertThat(processedProps).isEqualTo(allProps);
}
@Test
@@ -72,28 +71,28 @@ class DefaultDestinationTopicProcessorTests extends DestinationTopicTests {
registerThirdTopicDestinations(context);
// then
assertTrue(context.destinationsByTopicMap.containsKey(FIRST_TOPIC));
assertThat(context.destinationsByTopicMap.containsKey(FIRST_TOPIC)).isTrue();
List<DestinationTopic> destinationTopicsForFirstTopic = context.destinationsByTopicMap.get(FIRST_TOPIC);
assertEquals(4, destinationTopicsForFirstTopic.size());
assertEquals(mainDestinationTopic, destinationTopicsForFirstTopic.get(0));
assertEquals(firstRetryDestinationTopic, destinationTopicsForFirstTopic.get(1));
assertEquals(secondRetryDestinationTopic, destinationTopicsForFirstTopic.get(2));
assertEquals(dltDestinationTopic, destinationTopicsForFirstTopic.get(3));
assertThat(destinationTopicsForFirstTopic.size()).isEqualTo(4);
assertThat(destinationTopicsForFirstTopic.get(0)).isEqualTo(mainDestinationTopic);
assertThat(destinationTopicsForFirstTopic.get(1)).isEqualTo(firstRetryDestinationTopic);
assertThat(destinationTopicsForFirstTopic.get(2)).isEqualTo(secondRetryDestinationTopic);
assertThat(destinationTopicsForFirstTopic.get(3)).isEqualTo(dltDestinationTopic);
assertTrue(context.destinationsByTopicMap.containsKey(SECOND_TOPIC));
assertThat(context.destinationsByTopicMap.containsKey(SECOND_TOPIC)).isTrue();
List<DestinationTopic> destinationTopicsForSecondTopic = context.destinationsByTopicMap.get(SECOND_TOPIC);
assertEquals(4, destinationTopicsForSecondTopic.size());
assertEquals(mainDestinationTopic2, destinationTopicsForSecondTopic.get(0));
assertEquals(firstRetryDestinationTopic2, destinationTopicsForSecondTopic.get(1));
assertEquals(secondRetryDestinationTopic2, destinationTopicsForSecondTopic.get(2));
assertEquals(dltDestinationTopic2, destinationTopicsForSecondTopic.get(3));
assertThat(destinationTopicsForSecondTopic.size()).isEqualTo(4);
assertThat(destinationTopicsForSecondTopic.get(0)).isEqualTo(mainDestinationTopic2);
assertThat(destinationTopicsForSecondTopic.get(1)).isEqualTo(firstRetryDestinationTopic2);
assertThat(destinationTopicsForSecondTopic.get(2)).isEqualTo(secondRetryDestinationTopic2);
assertThat(destinationTopicsForSecondTopic.get(3)).isEqualTo(dltDestinationTopic2);
assertTrue(context.destinationsByTopicMap.containsKey(THIRD_TOPIC));
assertThat(context.destinationsByTopicMap.containsKey(THIRD_TOPIC)).isTrue();
List<DestinationTopic> destinationTopicsForThirdTopic = context.destinationsByTopicMap.get(THIRD_TOPIC);
assertEquals(3, destinationTopicsForThirdTopic.size());
assertEquals(mainDestinationTopic3, destinationTopicsForThirdTopic.get(0));
assertEquals(firstRetryDestinationTopic3, destinationTopicsForThirdTopic.get(1));
assertEquals(secondRetryDestinationTopic3, destinationTopicsForThirdTopic.get(2));
assertThat(destinationTopicsForThirdTopic.size()).isEqualTo(3);
assertThat(destinationTopicsForThirdTopic.get(0)).isEqualTo(mainDestinationTopic3);
assertThat(destinationTopicsForThirdTopic.get(1)).isEqualTo(firstRetryDestinationTopic3);
assertThat(destinationTopicsForThirdTopic.get(2)).isEqualTo(secondRetryDestinationTopic3);
}
private void registerFirstTopicDestinations(DestinationTopicProcessor.Context context) {
@@ -136,43 +135,43 @@ class DefaultDestinationTopicProcessorTests extends DestinationTopicTests {
then(destinationTopicResolver).should().addDestinations(destinationMapCaptor.capture());
Map<String, DestinationTopicResolver.DestinationsHolder> destinationMap = destinationMapCaptor.getValue();
assertEquals(11, destinationMap.size());
assertThat(destinationMap.size()).isEqualTo(11);
assertTrue(destinationMap.containsKey(mainDestinationTopic.getDestinationName()));
assertEquals(mainDestinationTopic, destinationMap.get(mainDestinationTopic.getDestinationName()).getSourceDestination());
assertEquals(firstRetryDestinationTopic, destinationMap.get(mainDestinationTopic.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(firstRetryDestinationTopic.getDestinationName()));
assertEquals(firstRetryDestinationTopic, destinationMap.get(firstRetryDestinationTopic.getDestinationName()).getSourceDestination());
assertEquals(secondRetryDestinationTopic, destinationMap.get(firstRetryDestinationTopic.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(secondRetryDestinationTopic.getDestinationName()));
assertEquals(secondRetryDestinationTopic, destinationMap.get(secondRetryDestinationTopic.getDestinationName()).getSourceDestination());
assertEquals(dltDestinationTopic, destinationMap.get(secondRetryDestinationTopic.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(dltDestinationTopic.getDestinationName()));
assertEquals(dltDestinationTopic, destinationMap.get(dltDestinationTopic.getDestinationName()).getSourceDestination());
assertEquals(noOpsDestinationTopic, destinationMap.get(dltDestinationTopic.getDestinationName()).getNextDestination());
assertThat(destinationMap.containsKey(mainDestinationTopic.getDestinationName())).isTrue();
assertThat(destinationMap.get(mainDestinationTopic.getDestinationName()).getSourceDestination()).isEqualTo(mainDestinationTopic);
assertThat(destinationMap.get(mainDestinationTopic.getDestinationName()).getNextDestination()).isEqualTo(firstRetryDestinationTopic);
assertThat(destinationMap.containsKey(firstRetryDestinationTopic.getDestinationName())).isTrue();
assertThat(destinationMap.get(firstRetryDestinationTopic.getDestinationName()).getSourceDestination()).isEqualTo(firstRetryDestinationTopic);
assertThat(destinationMap.get(firstRetryDestinationTopic.getDestinationName()).getNextDestination()).isEqualTo(secondRetryDestinationTopic);
assertThat(destinationMap.containsKey(secondRetryDestinationTopic.getDestinationName())).isTrue();
assertThat(destinationMap.get(secondRetryDestinationTopic.getDestinationName()).getSourceDestination()).isEqualTo(secondRetryDestinationTopic);
assertThat(destinationMap.get(secondRetryDestinationTopic.getDestinationName()).getNextDestination()).isEqualTo(dltDestinationTopic);
assertThat(destinationMap.containsKey(dltDestinationTopic.getDestinationName())).isTrue();
assertThat(destinationMap.get(dltDestinationTopic.getDestinationName()).getSourceDestination()).isEqualTo(dltDestinationTopic);
assertThat(destinationMap.get(dltDestinationTopic.getDestinationName()).getNextDestination()).isEqualTo(noOpsDestinationTopic);
assertTrue(destinationMap.containsKey(mainDestinationTopic2.getDestinationName()));
assertEquals(mainDestinationTopic2, destinationMap.get(mainDestinationTopic2.getDestinationName()).getSourceDestination());
assertEquals(firstRetryDestinationTopic2, destinationMap.get(mainDestinationTopic2.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(firstRetryDestinationTopic2.getDestinationName()));
assertEquals(firstRetryDestinationTopic2, destinationMap.get(firstRetryDestinationTopic2.getDestinationName()).getSourceDestination());
assertEquals(secondRetryDestinationTopic2, destinationMap.get(firstRetryDestinationTopic2.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(secondRetryDestinationTopic2.getDestinationName()));
assertEquals(secondRetryDestinationTopic2, destinationMap.get(secondRetryDestinationTopic2.getDestinationName()).getSourceDestination());
assertEquals(dltDestinationTopic2, destinationMap.get(secondRetryDestinationTopic2.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(dltDestinationTopic2.getDestinationName()));
assertEquals(dltDestinationTopic2, destinationMap.get(dltDestinationTopic2.getDestinationName()).getSourceDestination());
assertEquals(noOpsDestinationTopic2, destinationMap.get(dltDestinationTopic2.getDestinationName()).getNextDestination());
assertThat(destinationMap.containsKey(mainDestinationTopic2.getDestinationName())).isTrue();
assertThat(destinationMap.get(mainDestinationTopic2.getDestinationName()).getSourceDestination()).isEqualTo(mainDestinationTopic2);
assertThat(destinationMap.get(mainDestinationTopic2.getDestinationName()).getNextDestination()).isEqualTo(firstRetryDestinationTopic2);
assertThat(destinationMap.containsKey(firstRetryDestinationTopic2.getDestinationName())).isTrue();
assertThat(destinationMap.get(firstRetryDestinationTopic2.getDestinationName()).getSourceDestination()).isEqualTo(firstRetryDestinationTopic2);
assertThat(destinationMap.get(firstRetryDestinationTopic2.getDestinationName()).getNextDestination()).isEqualTo(secondRetryDestinationTopic2);
assertThat(destinationMap.containsKey(secondRetryDestinationTopic2.getDestinationName())).isTrue();
assertThat(destinationMap.get(secondRetryDestinationTopic2.getDestinationName()).getSourceDestination()).isEqualTo(secondRetryDestinationTopic2);
assertThat(destinationMap.get(secondRetryDestinationTopic2.getDestinationName()).getNextDestination()).isEqualTo(dltDestinationTopic2);
assertThat(destinationMap.containsKey(dltDestinationTopic2.getDestinationName())).isTrue();
assertThat(destinationMap.get(dltDestinationTopic2.getDestinationName()).getSourceDestination()).isEqualTo(dltDestinationTopic2);
assertThat(destinationMap.get(dltDestinationTopic2.getDestinationName()).getNextDestination()).isEqualTo(noOpsDestinationTopic2);
assertTrue(destinationMap.containsKey(mainDestinationTopic3.getDestinationName()));
assertEquals(mainDestinationTopic3, destinationMap.get(mainDestinationTopic3.getDestinationName()).getSourceDestination());
assertEquals(firstRetryDestinationTopic3, destinationMap.get(mainDestinationTopic3.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(firstRetryDestinationTopic3.getDestinationName()));
assertEquals(firstRetryDestinationTopic3, destinationMap.get(firstRetryDestinationTopic3.getDestinationName()).getSourceDestination());
assertEquals(secondRetryDestinationTopic3, destinationMap.get(firstRetryDestinationTopic3.getDestinationName()).getNextDestination());
assertTrue(destinationMap.containsKey(secondRetryDestinationTopic3.getDestinationName()));
assertEquals(secondRetryDestinationTopic3, destinationMap.get(secondRetryDestinationTopic3.getDestinationName()).getSourceDestination());
assertEquals(noOpsDestinationTopic3, destinationMap.get(secondRetryDestinationTopic3.getDestinationName()).getNextDestination());
assertThat(destinationMap.containsKey(mainDestinationTopic3.getDestinationName())).isTrue();
assertThat(destinationMap.get(mainDestinationTopic3.getDestinationName()).getSourceDestination()).isEqualTo(mainDestinationTopic3);
assertThat(destinationMap.get(mainDestinationTopic3.getDestinationName()).getNextDestination()).isEqualTo(firstRetryDestinationTopic3);
assertThat(destinationMap.containsKey(firstRetryDestinationTopic3.getDestinationName())).isTrue();
assertThat(destinationMap.get(firstRetryDestinationTopic3.getDestinationName()).getSourceDestination()).isEqualTo(firstRetryDestinationTopic3);
assertThat(destinationMap.get(firstRetryDestinationTopic3.getDestinationName()).getNextDestination()).isEqualTo(secondRetryDestinationTopic3);
assertThat(destinationMap.containsKey(secondRetryDestinationTopic3.getDestinationName())).isTrue();
assertThat(destinationMap.get(secondRetryDestinationTopic3.getDestinationName()).getSourceDestination()).isEqualTo(secondRetryDestinationTopic3);
assertThat(destinationMap.get(secondRetryDestinationTopic3.getDestinationName()).getNextDestination()).isEqualTo(noOpsDestinationTopic3);
}
@Test
@@ -201,7 +200,7 @@ class DefaultDestinationTopicProcessorTests extends DestinationTopicTests {
destinationTopicProcessor.processRegisteredDestinations(topics -> allProcessedTopics.addAll(topics), context);
// then
assertEquals(allTopics, allProcessedTopics);
assertThat(allProcessedTopics).isEqualTo(allTopics);
}
}

90
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicContainerTests.java
View File

@@ -16,8 +16,9 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.assertj.core.api.Assertions.assertThatNullPointerException;
import java.math.BigInteger;
import java.time.Clock;
@@ -91,99 +92,98 @@ class DestinationTopicContainerTests extends DestinationTopicTests {
@Test
void shouldResolveRetryDestination() {
assertEquals(firstRetryDestinationTopic, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(mainDestinationTopic.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
assertEquals(secondRetryDestinationTopic, destinationTopicContainer
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(firstRetryDestinationTopic);
assertThat(destinationTopicContainer
.resolveNextDestination(firstRetryDestinationTopic.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
assertEquals(dltDestinationTopic, destinationTopicContainer
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(secondRetryDestinationTopic);
assertThat(destinationTopicContainer
.resolveNextDestination(secondRetryDestinationTopic.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
assertEquals(noOpsDestinationTopic, destinationTopicContainer
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(dltDestinationTopic);
assertThat(destinationTopicContainer
.resolveNextDestination(dltDestinationTopic.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(noOpsDestinationTopic);
assertEquals(firstRetryDestinationTopic2, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(mainDestinationTopic2.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
assertEquals(secondRetryDestinationTopic2, destinationTopicContainer
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(firstRetryDestinationTopic2);
assertThat(destinationTopicContainer
.resolveNextDestination(firstRetryDestinationTopic2.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
assertEquals(dltDestinationTopic2, destinationTopicContainer
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(secondRetryDestinationTopic2);
assertThat(destinationTopicContainer
.resolveNextDestination(secondRetryDestinationTopic2.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
assertEquals(dltDestinationTopic2, destinationTopicContainer
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(dltDestinationTopic2);
assertThat(destinationTopicContainer
.resolveNextDestination(dltDestinationTopic2.getDestinationName(), 1,
new IllegalArgumentException(), this.originalTimestamp));
new IllegalArgumentException(), this.originalTimestamp)).isEqualTo(dltDestinationTopic2);
}
@Test
void shouldResolveDltDestinationForNonRetryableException() {
assertEquals(dltDestinationTopic, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(mainDestinationTopic.getDestinationName(),
1, new RuntimeException(), originalTimestamp));
1, new RuntimeException(), originalTimestamp)).isEqualTo(dltDestinationTopic);
}
@Test
void shouldResolveRetryDestinationForWrappedException() {
assertEquals(firstRetryDestinationTopic, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(mainDestinationTopic.getDestinationName(),
1, new ListenerExecutionFailedException("Test exception!",
new IllegalArgumentException()), originalTimestamp));
new IllegalArgumentException()), originalTimestamp)).isEqualTo(firstRetryDestinationTopic);
}
@Test
void shouldResolveNoOpsDestinationForDoNotRetryDltPolicy() {
assertEquals(noOpsDestinationTopic, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(dltDestinationTopic.getDestinationName(),
1, new IllegalArgumentException(), originalTimestamp));
1, new IllegalArgumentException(), originalTimestamp)).isEqualTo(noOpsDestinationTopic);
}
@Test
void shouldResolveDltDestinationForAlwaysRetryDltPolicy() {
assertEquals(dltDestinationTopic2, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(dltDestinationTopic2.getDestinationName(),
1, new IllegalArgumentException(), originalTimestamp));
1, new IllegalArgumentException(), originalTimestamp)).isEqualTo(dltDestinationTopic2);
}
@Test
void shouldResolveDltDestinationForExpiredTimeout() {
long timestampInThePastToForceTimeout = this.originalTimestamp - 10000;
assertEquals(dltDestinationTopic2, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(mainDestinationTopic2.getDestinationName(),
1, new IllegalArgumentException(), timestampInThePastToForceTimeout));
1, new IllegalArgumentException(), timestampInThePastToForceTimeout)).isEqualTo(dltDestinationTopic2);
}
@Test
void shouldThrowIfNoDestinationFound() {
assertThrows(NullPointerException.class,
() -> destinationTopicContainer.resolveNextDestination("Non-existing-topic", 0,
assertThatNullPointerException().isThrownBy(() -> destinationTopicContainer.resolveNextDestination("Non-existing-topic", 0,
new IllegalArgumentException(), originalTimestamp));
}
@Test
void shouldResolveNoOpsIfDltAndNotRetryable() {
assertEquals(noOpsDestinationTopic3, destinationTopicContainer
assertThat(destinationTopicContainer
.resolveNextDestination(mainDestinationTopic3.getDestinationName(), 0,
new RuntimeException(), originalTimestamp));
new RuntimeException(), originalTimestamp)).isEqualTo(noOpsDestinationTopic3);
}
@Test
void shouldResolveDestinationNextExecutionTime() {
RuntimeException e = new IllegalArgumentException();
assertEquals(getExpectedNextExecutionTime(firstRetryDestinationTopic),
destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
mainDestinationTopic.getDestinationName(), 0, e, originalTimestamp));
assertEquals(getExpectedNextExecutionTime(secondRetryDestinationTopic),
destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
firstRetryDestinationTopic.getDestinationName(), 0, e, originalTimestamp));
assertEquals(getExpectedNextExecutionTime(dltDestinationTopic),
destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
secondRetryDestinationTopic.getDestinationName(), 0, e, originalTimestamp));
assertEquals(getExpectedNextExecutionTime(noOpsDestinationTopic),
destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
dltDestinationTopic.getDestinationName(), 0, e, originalTimestamp));
assertThat(destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
mainDestinationTopic.getDestinationName(), 0, e, originalTimestamp))
.isEqualTo(getExpectedNextExecutionTime(firstRetryDestinationTopic));
assertThat(destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
firstRetryDestinationTopic.getDestinationName(), 0, e, originalTimestamp))
.isEqualTo(getExpectedNextExecutionTime(secondRetryDestinationTopic));
assertThat(destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
secondRetryDestinationTopic.getDestinationName(), 0, e, originalTimestamp))
.isEqualTo(getExpectedNextExecutionTime(dltDestinationTopic));
assertThat(destinationTopicContainer.resolveDestinationNextExecutionTimestamp(
dltDestinationTopic.getDestinationName(), 0, e, originalTimestamp))
.isEqualTo(getExpectedNextExecutionTime(noOpsDestinationTopic));
}
private long getExpectedNextExecutionTime(DestinationTopic destinationTopic) {
@@ -193,7 +193,7 @@ class DestinationTopicContainerTests extends DestinationTopicTests {
@Test
void shouldThrowIfAddsDestinationsAfterClosed() {
destinationTopicContainer.onApplicationEvent(null);
assertThrows(IllegalStateException.class, () ->
assertThatIllegalStateException().isThrownBy(() ->
destinationTopicContainer.addDestinations(Collections.emptyMap()));
}
}

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

@@ -16,9 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.List;
import java.util.stream.Collectors;
@@ -91,30 +89,30 @@ class DestinationTopicPropertiesFactoryTests {
.createProperties();
// then
assertTrue(propertiesList.size() == 2);
assertThat(propertiesList.size() == 2).isTrue();
DestinationTopic.Properties mainTopicProperties = propertiesList.get(0);
assertEquals("", mainTopicProperties.suffix());
assertFalse(mainTopicProperties.isDltTopic());
assertThat(mainTopicProperties.suffix()).isEqualTo("");
assertThat(mainTopicProperties.isDltTopic()).isFalse();
DestinationTopic mainTopic = new DestinationTopic("mainTopic", mainTopicProperties);
assertEquals(0L, mainTopic.getDestinationDelay());
assertTrue(mainTopic.shouldRetryOn(0, new IllegalArgumentException()));
assertFalse(mainTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException()));
assertFalse(mainTopic.shouldRetryOn(0, new RuntimeException()));
assertEquals(RetryTopicConstants.NOT_SET, mainTopic.getDestinationTimeout());
assertThat(mainTopic.getDestinationDelay()).isEqualTo(0L);
assertThat(mainTopic.shouldRetryOn(0, new IllegalArgumentException())).isTrue();
assertThat(mainTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException())).isFalse();
assertThat(mainTopic.shouldRetryOn(0, new RuntimeException())).isFalse();
assertThat(mainTopic.getDestinationTimeout()).isEqualTo(RetryTopicConstants.NOT_SET);
DestinationTopic.Properties dltProperties = propertiesList.get(1);
assertDltTopic(dltProperties);
}
private void assertDltTopic(DestinationTopic.Properties dltProperties) {
assertEquals(dltSuffix, dltProperties.suffix());
assertTrue(dltProperties.isDltTopic());
assertThat(dltProperties.suffix()).isEqualTo(dltSuffix);
assertThat(dltProperties.isDltTopic()).isTrue();
DestinationTopic dltTopic = new DestinationTopic("mainTopic", dltProperties);
assertEquals(0, dltTopic.getDestinationDelay());
assertFalse(dltTopic.shouldRetryOn(0, new IllegalArgumentException()));
assertFalse(dltTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException()));
assertFalse(dltTopic.shouldRetryOn(0, new RuntimeException()));
assertEquals(RetryTopicConstants.NOT_SET, dltTopic.getDestinationTimeout());
assertThat(dltTopic.getDestinationDelay()).isEqualTo(0);
assertThat(dltTopic.shouldRetryOn(0, new IllegalArgumentException())).isFalse();
assertThat(dltTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException())).isFalse();
assertThat(dltTopic.shouldRetryOn(0, new RuntimeException())).isFalse();
assertThat(dltTopic.getDestinationTimeout()).isEqualTo(RetryTopicConstants.NOT_SET);
}
@Test
@@ -137,26 +135,26 @@ class DestinationTopicPropertiesFactoryTests {
.collect(Collectors.toList());
// then
assertTrue(propertiesList.size() == 4);
assertThat(propertiesList.size() == 4).isTrue();
DestinationTopic.Properties firstRetryProperties = propertiesList.get(1);
assertEquals(retryTopicSuffix + "-1000", firstRetryProperties.suffix());
assertFalse(firstRetryProperties.isDltTopic());
assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix + "-1000");
assertThat(firstRetryProperties.isDltTopic()).isFalse();
DestinationTopic firstRetryDestinationTopic = destinationTopicList.get(1);
assertEquals(1000, firstRetryDestinationTopic.getDestinationDelay());
assertEquals(numPartitions, firstRetryDestinationTopic.getDestinationPartitions());
assertTrue(firstRetryDestinationTopic.shouldRetryOn(0, new IllegalArgumentException()));
assertFalse(firstRetryDestinationTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException()));
assertFalse(firstRetryDestinationTopic.shouldRetryOn(0, new RuntimeException()));
assertThat(firstRetryDestinationTopic.getDestinationDelay()).isEqualTo(1000);
assertThat(firstRetryDestinationTopic.getDestinationPartitions()).isEqualTo(numPartitions);
assertThat(firstRetryDestinationTopic.shouldRetryOn(0, new IllegalArgumentException())).isTrue();
assertThat(firstRetryDestinationTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException())).isFalse();
assertThat(firstRetryDestinationTopic.shouldRetryOn(0, new RuntimeException())).isFalse();
DestinationTopic.Properties secondRetryProperties = propertiesList.get(2);
assertEquals(retryTopicSuffix + "-2000", secondRetryProperties.suffix());
assertFalse(secondRetryProperties.isDltTopic());
assertThat(secondRetryProperties.suffix()).isEqualTo(retryTopicSuffix + "-2000");
assertThat(secondRetryProperties.isDltTopic()).isFalse();
DestinationTopic secondRetryDestinationTopic = destinationTopicList.get(2);
assertEquals(2000, secondRetryDestinationTopic.getDestinationDelay());
assertEquals(numPartitions, secondRetryDestinationTopic.getDestinationPartitions());
assertTrue(secondRetryDestinationTopic.shouldRetryOn(0, new IllegalArgumentException()));
assertFalse(secondRetryDestinationTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException()));
assertFalse(secondRetryDestinationTopic.shouldRetryOn(0, new RuntimeException()));
assertThat(secondRetryDestinationTopic.getDestinationDelay()).isEqualTo(2000);
assertThat(secondRetryDestinationTopic.getDestinationPartitions()).isEqualTo(numPartitions);
assertThat(secondRetryDestinationTopic.shouldRetryOn(0, new IllegalArgumentException())).isTrue();
assertThat(secondRetryDestinationTopic.shouldRetryOn(maxAttempts, new IllegalArgumentException())).isFalse();
assertThat(secondRetryDestinationTopic.shouldRetryOn(0, new RuntimeException())).isFalse();
assertDltTopic(propertiesList.get(3));
}
@@ -182,8 +180,8 @@ class DestinationTopicPropertiesFactoryTests {
.collect(Collectors.toList());
// then
assertTrue(propertiesList.size() == 3);
assertFalse(propertiesList.get(2).isDltTopic());
assertThat(propertiesList.size() == 3).isTrue();
assertThat(propertiesList.get(2).isDltTopic()).isFalse();
}
@Test
@@ -205,24 +203,24 @@ class DestinationTopicPropertiesFactoryTests {
.collect(Collectors.toList());
// then
assertTrue(propertiesList.size() == 3);
assertThat(propertiesList.size() == 3).isTrue();
DestinationTopic.Properties mainTopicProperties = propertiesList.get(0);
DestinationTopic mainDestinationTopic = destinationTopicList.get(0);
assertTrue(mainDestinationTopic.isMainTopic());
assertThat(mainDestinationTopic.isMainTopic()).isTrue();
DestinationTopic.Properties firstRetryProperties = propertiesList.get(1);
assertEquals(retryTopicSuffix, firstRetryProperties.suffix());
assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix);
DestinationTopic retryDestinationTopic = destinationTopicList.get(1);
assertTrue(retryDestinationTopic.isSingleTopicRetry());
assertEquals(1000, retryDestinationTopic.getDestinationDelay());
assertThat(retryDestinationTopic.isSingleTopicRetry()).isTrue();
assertThat(retryDestinationTopic.getDestinationDelay()).isEqualTo(1000);
DestinationTopic.Properties dltProperties = propertiesList.get(2);
assertEquals(dltSuffix, dltProperties.suffix());
assertTrue(dltProperties.isDltTopic());
assertThat(dltProperties.suffix()).isEqualTo(dltSuffix);
assertThat(dltProperties.isDltTopic()).isTrue();
DestinationTopic dltTopic = destinationTopicList.get(2);
assertEquals(0, dltTopic.getDestinationDelay());
assertEquals(numPartitions, dltTopic.getDestinationPartitions());
assertThat(dltTopic.getDestinationDelay()).isEqualTo(0);
assertThat(dltTopic.getDestinationPartitions()).isEqualTo(numPartitions);
}
@Test
@@ -245,30 +243,30 @@ class DestinationTopicPropertiesFactoryTests {
.collect(Collectors.toList());
// then
assertTrue(propertiesList.size() == 4);
assertThat(propertiesList.size() == 4).isTrue();
DestinationTopic.Properties mainTopicProperties = propertiesList.get(0);
DestinationTopic mainDestinationTopic = destinationTopicList.get(0);
assertTrue(mainDestinationTopic.isMainTopic());
assertThat(mainDestinationTopic.isMainTopic()).isTrue();
DestinationTopic.Properties firstRetryProperties = propertiesList.get(1);
assertEquals(retryTopicSuffix + "-0", firstRetryProperties.suffix());
assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix + "-0");
DestinationTopic retryDestinationTopic = destinationTopicList.get(1);
assertFalse(retryDestinationTopic.isSingleTopicRetry());
assertEquals(5000, retryDestinationTopic.getDestinationDelay());
assertThat(retryDestinationTopic.isSingleTopicRetry()).isFalse();
assertThat(retryDestinationTopic.getDestinationDelay()).isEqualTo(5000);
DestinationTopic.Properties secondRetryProperties = propertiesList.get(2);
assertEquals(retryTopicSuffix + "-1", secondRetryProperties.suffix());
assertThat(secondRetryProperties.suffix()).isEqualTo(retryTopicSuffix + "-1");
DestinationTopic secondRetryDestinationTopic = destinationTopicList.get(2);
assertFalse(secondRetryDestinationTopic.isSingleTopicRetry());
assertEquals(5000, secondRetryDestinationTopic.getDestinationDelay());
assertThat(secondRetryDestinationTopic.isSingleTopicRetry()).isFalse();
assertThat(secondRetryDestinationTopic.getDestinationDelay()).isEqualTo(5000);
DestinationTopic.Properties dltProperties = propertiesList.get(3);
assertEquals(dltSuffix, dltProperties.suffix());
assertTrue(dltProperties.isDltTopic());
assertThat(dltProperties.suffix()).isEqualTo(dltSuffix);
assertThat(dltProperties.isDltTopic()).isTrue();
DestinationTopic dltTopic = destinationTopicList.get(3);
assertEquals(0, dltTopic.getDestinationDelay());
assertEquals(numPartitions, dltTopic.getDestinationPartitions());
assertThat(dltTopic.getDestinationDelay()).isEqualTo(0);
assertThat(dltTopic.getDestinationPartitions()).isEqualTo(numPartitions);
}
@Test
@@ -287,8 +285,7 @@ class DestinationTopicPropertiesFactoryTests {
// then
IntStream.range(1, maxAttempts)
.forEach(index -> assertEquals(retryTopicSuffix + "-" + String.valueOf(index - 1),
propertiesList.get(index).suffix()));
.forEach(index -> assertThat(propertiesList.get(index).suffix()).isEqualTo(retryTopicSuffix + "-" + String.valueOf(index - 1)));
}
@Test
@@ -308,8 +305,7 @@ class DestinationTopicPropertiesFactoryTests {
// then
IntStream.range(1, maxAttempts)
.forEach(index -> assertEquals(retryTopicSuffix + "-" + String.valueOf(index - 1),
propertiesList.get(index).suffix()));
.forEach(index -> assertThat(propertiesList.get(index).suffix()).isEqualTo(retryTopicSuffix + "-" + String.valueOf(index - 1)));
}
@Test
@@ -330,12 +326,12 @@ class DestinationTopicPropertiesFactoryTests {
dltStrategy, defaultTopicSuffixingStrategy, -1).createProperties();
// then
assertTrue(propertiesList.size() == 6);
assertEquals("", propertiesList.get(0).suffix());
assertEquals(retryTopicSuffix + "-1000", propertiesList.get(1).suffix());
assertEquals(retryTopicSuffix + "-2000", propertiesList.get(2).suffix());
assertEquals(retryTopicSuffix + "-3000-0", propertiesList.get(3).suffix());
assertEquals(retryTopicSuffix + "-3000-1", propertiesList.get(4).suffix());
assertEquals(dltSuffix, propertiesList.get(5).suffix());
assertThat(propertiesList.size() == 6).isTrue();
assertThat(propertiesList.get(0).suffix()).isEqualTo("");
assertThat(propertiesList.get(1).suffix()).isEqualTo(retryTopicSuffix + "-1000");
assertThat(propertiesList.get(2).suffix()).isEqualTo(retryTopicSuffix + "-2000");
assertThat(propertiesList.get(3).suffix()).isEqualTo(retryTopicSuffix + "-3000-0");
assertThat(propertiesList.get(4).suffix()).isEqualTo(retryTopicSuffix + "-3000-1");
assertThat(propertiesList.get(5).suffix()).isEqualTo(dltSuffix);
}
}

9
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java
View File

@@ -16,8 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
@@ -149,7 +148,7 @@ class ListenerContainerFactoryConfigurerTests {
// then
then(containerFactory).should(times(1)).setErrorHandler(errorHandlerCaptor.capture());
ErrorHandler errorHandler = errorHandlerCaptor.getValue();
assertTrue(SeekToCurrentErrorHandler.class.isAssignableFrom(errorHandler.getClass()));
assertThat(SeekToCurrentErrorHandler.class.isAssignableFrom(errorHandler.getClass())).isTrue();
SeekToCurrentErrorHandler seekToCurrent = (SeekToCurrentErrorHandler) errorHandler;
RuntimeException ex = new RuntimeException();
@@ -250,7 +249,7 @@ class ListenerContainerFactoryConfigurerTests {
then(this.kafkaConsumerBackoffManager).should(times(1))
.createContext(anyLong(), listenerIdCaptor.capture(), any(TopicPartition.class));
assertEquals(testListenerId, listenerIdCaptor.getValue());
assertThat(listenerIdCaptor.getValue()).isEqualTo(testListenerId);
then(listener).should(times(1)).onMessage(data, ack, consumer);
then(this.configurerContainerCustomizer).should(times(1)).accept(container);
@@ -272,7 +271,7 @@ class ListenerContainerFactoryConfigurerTests {
.configure(containerFactory);
// then
assertEquals(factory, secondFactory);
assertThat(secondFactory).isEqualTo(factory);
then(containerFactory).should(times(1)).setContainerCustomizer(any());
then(containerFactory).should(times(1)).setErrorHandler(any());
}

52
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryResolverTests.java
View File

@@ -16,9 +16,8 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.times;
@@ -73,7 +72,7 @@ class ListenerContainerFactoryResolverTests {
.resolveFactoryForMainEndpoint(factoryFromKafkaListenerAnnotation, defaultFactoryBeanName, configuration);
// then
assertEquals(factoryFromKafkaListenerAnnotation, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromKafkaListenerAnnotation);
}
@Test
@@ -89,7 +88,7 @@ class ListenerContainerFactoryResolverTests {
.resolveFactoryForMainEndpoint(null, defaultFactoryBeanName, configuration);
// then
assertEquals(factoryFromRetryTopicConfiguration, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromRetryTopicConfiguration);
}
@Test
@@ -107,7 +106,7 @@ class ListenerContainerFactoryResolverTests {
.resolveFactoryForMainEndpoint(null, null, configuration);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
}
@Test
@@ -125,7 +124,7 @@ class ListenerContainerFactoryResolverTests {
defaultFactoryBeanName, configuration);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
}
@Test
@@ -143,7 +142,7 @@ class ListenerContainerFactoryResolverTests {
listenerContainerFactoryResolver.resolveFactoryForMainEndpoint(null, null, configuration);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
}
@Test
@@ -157,9 +156,8 @@ class ListenerContainerFactoryResolverTests {
new ListenerContainerFactoryResolver.Configuration(null, null);
// then
assertThrows(IllegalArgumentException.class,
() -> listenerContainerFactoryResolver
.resolveFactoryForMainEndpoint(null, null, configuration));
assertThatExceptionOfType(IllegalArgumentException.class).isThrownBy(() -> listenerContainerFactoryResolver
.resolveFactoryForMainEndpoint(null, null, configuration));
}
@Test
@@ -175,7 +173,7 @@ class ListenerContainerFactoryResolverTests {
.resolveFactoryForRetryEndpoint(factoryFromKafkaListenerAnnotation, null, configuration);
// then
assertEquals(factoryFromRetryTopicConfiguration, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromRetryTopicConfiguration);
}
@Test
@@ -192,7 +190,7 @@ class ListenerContainerFactoryResolverTests {
.resolveFactoryForRetryEndpoint(factoryFromKafkaListenerAnnotation, null, configuration);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
}
@Test
@@ -208,7 +206,7 @@ class ListenerContainerFactoryResolverTests {
listenerContainerFactoryResolver.resolveFactoryForRetryEndpoint(factoryFromKafkaListenerAnnotation, null, configuration);
// then
assertEquals(factoryFromKafkaListenerAnnotation, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromKafkaListenerAnnotation);
}
@Test
@@ -227,7 +225,7 @@ class ListenerContainerFactoryResolverTests {
defaultFactoryBeanName, configuration);
// then
assertEquals(factoryFromDefaultBeanName, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromDefaultBeanName);
}
@Test
@@ -246,7 +244,7 @@ class ListenerContainerFactoryResolverTests {
defaultFactoryBeanName, configuration);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
}
@Test
@@ -260,8 +258,8 @@ class ListenerContainerFactoryResolverTests {
new ListenerContainerFactoryResolver.Configuration(null, null);
// then
assertThrows(IllegalArgumentException.class,
() -> listenerContainerFactoryResolver.resolveFactoryForRetryEndpoint(null, defaultFactoryBeanName, configuration));
assertThatExceptionOfType(IllegalArgumentException.class)
.isThrownBy(() -> listenerContainerFactoryResolver.resolveFactoryForRetryEndpoint(null, defaultFactoryBeanName, configuration));
}
@Test
@@ -288,8 +286,8 @@ class ListenerContainerFactoryResolverTests {
null, configuration2);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertEquals(factoryFromBeanName, resolvedFactory2);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
assertThat(resolvedFactory2).isEqualTo(factoryFromBeanName);
then(beanFactory).should(times(1)).getBean(factoryName,
ConcurrentKafkaListenerContainerFactory.class);
}
@@ -318,8 +316,8 @@ class ListenerContainerFactoryResolverTests {
null, configuration2);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertEquals(factoryFromBeanName, resolvedFactory2);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
assertThat(resolvedFactory2).isEqualTo(factoryFromBeanName);
then(beanFactory).should(times(1)).getBean(factoryName,
ConcurrentKafkaListenerContainerFactory.class);
}
@@ -350,8 +348,8 @@ class ListenerContainerFactoryResolverTests {
null, configuration2);
// then
assertEquals(factoryFromBeanName, resolvedFactory);
assertEquals(factoryFromOtherBeanName, resolvedFactory2);
assertThat(resolvedFactory).isEqualTo(factoryFromBeanName);
assertThat(resolvedFactory2).isEqualTo(factoryFromOtherBeanName);
then(beanFactory).should(times(1)).getBean(factoryName,
ConcurrentKafkaListenerContainerFactory.class);
then(beanFactory).should(times(1)).getBean(otherFactoryName,
@@ -370,7 +368,7 @@ class ListenerContainerFactoryResolverTests {
cache.addIfAbsent(factoryFromKafkaListenerAnnotation, configuration, factoryFromDefaultBeanName);
// then
assertEquals(cache.fromCache(factoryFromKafkaListenerAnnotation, configuration), factoryFromDefaultBeanName);
assertThat(factoryFromDefaultBeanName).isEqualTo(cache.fromCache(factoryFromKafkaListenerAnnotation, configuration));
}
@Test
@@ -387,7 +385,7 @@ class ListenerContainerFactoryResolverTests {
cache.addIfAbsent(factoryFromKafkaListenerAnnotation, configuration, factoryFromDefaultBeanName);
// then
assertEquals(cache.fromCache(factoryFromKafkaListenerAnnotation, configuration2), factoryFromDefaultBeanName);
assertThat(factoryFromDefaultBeanName).isEqualTo(cache.fromCache(factoryFromKafkaListenerAnnotation, configuration2));
}
@Test
@@ -404,6 +402,6 @@ class ListenerContainerFactoryResolverTests {
cache.addIfAbsent(factoryFromKafkaListenerAnnotation, configuration, factoryFromDefaultBeanName);
// then
assertNull(cache.fromCache(factoryFromKafkaListenerAnnotation, configuration2));
assertThat(cache.fromCache(factoryFromKafkaListenerAnnotation, configuration2)).isNull();
}
}

8
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapperTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
@@ -66,12 +66,14 @@ class RetryTopicBootstrapperTests {
@Test
void shouldThrowIfACDoesntImplementInterfaces() {
assertThrows(IllegalStateException.class, () -> new RetryTopicBootstrapper(wrongApplicationContext, beanFactory));
assertThatIllegalStateException()
.isThrownBy(() -> new RetryTopicBootstrapper(wrongApplicationContext, beanFactory));
}
@Test
void shouldThrowIfBFDoesntImplementInterfaces() {
assertThrows(IllegalStateException.class, () -> new RetryTopicBootstrapper(applicationContext, wrongBeanFactory));
assertThatIllegalStateException()
.isThrownBy(() -> new RetryTopicBootstrapper(applicationContext, wrongBeanFactory));
}
@Test

50
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilderTests.java
View File

@@ -16,9 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.List;
@@ -60,7 +58,7 @@ class RetryTopicConfigurationBuilderTests {
RetryTopicConfiguration configuration = builder.create(kafkaOperations);
// then
assertFalse(configuration.hasConfigurationForTopics(topicNames));
assertThat(configuration.hasConfigurationForTopics(topicNames)).isFalse();
}
@Test
@@ -75,10 +73,10 @@ class RetryTopicConfigurationBuilderTests {
// then
List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
assertEquals(0, destinationTopicProperties.get(0).delay());
assertEquals(1000, destinationTopicProperties.get(1).delay());
assertEquals(1000, destinationTopicProperties.get(2).delay());
assertEquals(0, destinationTopicProperties.get(3).delay());
assertThat(destinationTopicProperties.get(0).delay()).isEqualTo(0);
assertThat(destinationTopicProperties.get(1).delay()).isEqualTo(1000);
assertThat(destinationTopicProperties.get(2).delay()).isEqualTo(1000);
assertThat(destinationTopicProperties.get(3).delay()).isEqualTo(0);
}
@Test
@@ -93,10 +91,12 @@ class RetryTopicConfigurationBuilderTests {
// then
List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
assertEquals(0, destinationTopicProperties.get(0).delay());
assertEquals(0, destinationTopicProperties.get(1).delay());
assertEquals(0, destinationTopicProperties.get(2).delay());
assertEquals(0, destinationTopicProperties.get(3).delay());
assertThat(destinationTopicProperties.get(0).delay()).isEqualTo(0);
assertThat(destinationTopicProperties.get(1).delay()).isEqualTo(0);
assertThat(destinationTopicProperties.get(2).delay()).isEqualTo(0);
assertThat(destinationTopicProperties.get(3).delay()).isEqualTo(0);
}
@Test
@@ -113,12 +113,12 @@ class RetryTopicConfigurationBuilderTests {
// then
List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
assertEquals(0, destinationTopicProperties.get(0).delay());
assertTrue(minInterval < destinationTopicProperties.get(1).delay());
assertTrue(destinationTopicProperties.get(1).delay() < maxInterval);
assertTrue(minInterval < destinationTopicProperties.get(2).delay());
assertTrue(destinationTopicProperties.get(2).delay() < maxInterval);
assertEquals(0, destinationTopicProperties.get(3).delay());
assertThat(destinationTopicProperties.get(0).delay()).isEqualTo(0);
assertThat(minInterval < destinationTopicProperties.get(1).delay()).isTrue();
assertThat(destinationTopicProperties.get(1).delay() < maxInterval).isTrue();
assertThat(minInterval < destinationTopicProperties.get(2).delay()).isTrue();
assertThat(destinationTopicProperties.get(2).delay() < maxInterval).isTrue();
assertThat(destinationTopicProperties.get(3).delay()).isEqualTo(0);
}
@Test
@@ -132,8 +132,8 @@ class RetryTopicConfigurationBuilderTests {
// then
DestinationTopic destinationTopic = new DestinationTopic("",
configuration.getDestinationTopicProperties().get(0));
assertTrue(destinationTopic.shouldRetryOn(0, new IllegalArgumentException()));
assertFalse(destinationTopic.shouldRetryOn(0, new IllegalStateException()));
assertThat(destinationTopic.shouldRetryOn(0, new IllegalArgumentException())).isTrue();
assertThat(destinationTopic.shouldRetryOn(0, new IllegalStateException())).isFalse();
}
@Test
@@ -148,8 +148,8 @@ class RetryTopicConfigurationBuilderTests {
// then
DestinationTopic destinationTopic = new DestinationTopic("",
configuration.getDestinationTopicProperties().get(0));
assertFalse(destinationTopic.shouldRetryOn(0, new IllegalArgumentException()));
assertTrue(destinationTopic.shouldRetryOn(0, new IllegalStateException()));
assertThat(destinationTopic.shouldRetryOn(0, new IllegalArgumentException())).isFalse();
assertThat(destinationTopic.shouldRetryOn(0, new IllegalStateException())).isTrue();
}
@Test
@@ -167,8 +167,8 @@ class RetryTopicConfigurationBuilderTests {
Object factoryInstance = ReflectionTestUtils.getField(config, "factoryFromRetryTopicConfiguration");
Object listenerContainerFactoryName = ReflectionTestUtils.getField(config, "listenerContainerFactoryName");
assertEquals(containerFactory, factoryInstance);
assertEquals(factoryName, listenerContainerFactoryName);
assertThat(factoryInstance).isEqualTo(containerFactory);
assertThat(listenerContainerFactoryName).isEqualTo(factoryName);
}
@@ -182,6 +182,6 @@ class RetryTopicConfigurationBuilderTests {
.create(kafkaOperations);
RetryTopicConfiguration.TopicCreation config = configuration.forKafkaTopicAutoCreation();
assertFalse(config.shouldCreateTopics());
assertThat(config.shouldCreateTopics()).isFalse();
}
}

9
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationProviderTests.java
View File

@@ -16,8 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
import static org.mockito.BDDMockito.willReturn;
@@ -105,7 +104,7 @@ class RetryTopicConfigurationProviderTests {
// then
then(this.beanFactory).should(times(1)).getBeansOfType(RetryTopicConfiguration.class);
assertEquals(retryTopicConfiguration, configuration);
assertThat(configuration).isEqualTo(retryTopicConfiguration);
}
@@ -123,7 +122,7 @@ class RetryTopicConfigurationProviderTests {
// then
then(this.beanFactory).should(times(1)).getBeansOfType(RetryTopicConfiguration.class);
assertNull(configuration);
assertThat(configuration).isNull();
}
@@ -136,7 +135,7 @@ class RetryTopicConfigurationProviderTests {
RetryTopicConfiguration configuration = provider.findRetryConfigurationFor(topics, nonAnnotatedMethod, bean);
// then
assertNull(configuration);
assertThat(configuration).isNull();
}

39
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java
View File

@@ -16,9 +16,8 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given;
@@ -173,7 +172,7 @@ class RetryTopicConfigurerTests {
listenerContainerFactoryConfigurer, beanFactory);
// when - then
assertThrows(IllegalArgumentException.class,
assertThatIllegalArgumentException().isThrownBy(
() -> configurer.processMainAndRetryListeners(endpointProcessor, multiMethodEndpoint, configuration,
registrar, containerFactory, defaultFactoryBeanName));
}
@@ -250,22 +249,22 @@ class RetryTopicConfigurerTests {
then(registrar).should(times(4)).registerEndpoint(endpointCaptor.capture(), eq(this.containerFactory));
List<MethodKafkaListenerEndpoint<?, ?>> allRegisteredEndpoints = endpointCaptor.getAllValues();
assertEquals(mainEndpoint, allRegisteredEndpoints.get(0));
assertThat(allRegisteredEndpoints.get(0)).isEqualTo(mainEndpoint);
List<String> firstRetryTopics = new ArrayList<>(allRegisteredEndpoints.get(1).getTopics());
List<String> secondRetryTopics = new ArrayList<>(allRegisteredEndpoints.get(2).getTopics());
List<String> dltTopics = new ArrayList<>(allRegisteredEndpoints.get(3).getTopics());
assertEquals(topics.get(0) + firstRetrySuffix, firstRetryTopics.get(0));
assertEquals(topics.get(1) + firstRetrySuffix, firstRetryTopics.get(1));
assertEquals(topics.get(0) + secondRetrySuffix, secondRetryTopics.get(0));
assertEquals(topics.get(1) + secondRetrySuffix, secondRetryTopics.get(1));
assertEquals(topics.get(0) + dltSuffix, dltTopics.get(0));
assertEquals(topics.get(1) + dltSuffix, dltTopics.get(1));
assertThat(firstRetryTopics.get(0)).isEqualTo(topics.get(0) + firstRetrySuffix);
assertThat(firstRetryTopics.get(1)).isEqualTo(topics.get(1) + firstRetrySuffix);
assertThat(secondRetryTopics.get(0)).isEqualTo(topics.get(0) + secondRetrySuffix);
assertThat(secondRetryTopics.get(1)).isEqualTo(topics.get(1) + secondRetrySuffix);
assertThat(dltTopics.get(0)).isEqualTo(topics.get(0) + dltSuffix);
assertThat(dltTopics.get(1)).isEqualTo(topics.get(1) + dltSuffix);
assertEquals(this.defaultListableBeanFactory, ReflectionTestUtils.getField(allRegisteredEndpoints.get(1), "beanFactory"));
assertEquals(this.defaultListableBeanFactory, ReflectionTestUtils.getField(allRegisteredEndpoints.get(2), "beanFactory"));
assertEquals(this.defaultListableBeanFactory, ReflectionTestUtils.getField(allRegisteredEndpoints.get(3), "beanFactory"));
assertThat(ReflectionTestUtils.getField(allRegisteredEndpoints.get(1), "beanFactory")).isEqualTo(this.defaultListableBeanFactory);
assertThat(ReflectionTestUtils.getField(allRegisteredEndpoints.get(2), "beanFactory")).isEqualTo(this.defaultListableBeanFactory);
assertThat(ReflectionTestUtils.getField(allRegisteredEndpoints.get(3), "beanFactory")).isEqualTo(this.defaultListableBeanFactory);
then(destinationTopicProcessor).should(times(1)).processRegisteredDestinations(topicsConsumerCaptor.capture(), eq(context));
@@ -285,10 +284,10 @@ class RetryTopicConfigurerTests {
List<String> allValues = mainTopicNameCaptor.getAllValues();
List<String> retryTopicName = retryDltTopicNameCaptor.getAllValues();
assertEquals(topics.get(0), allValues.get(index));
assertEquals(topics.get(1), allValues.get(index + 1));
assertEquals(firstTopicName, retryTopicName.get(index));
assertEquals(secondTopicName, retryTopicName.get(index + 1));
assertThat(allValues.get(index)).isEqualTo(topics.get(0));
assertThat(allValues.get(index + 1)).isEqualTo(topics.get(1));
assertThat(retryTopicName.get(index)).isEqualTo(firstTopicName);
assertThat(retryTopicName.get(index + 1)).isEqualTo(secondTopicName);
}
private void thenAssertEndpointProcessing(MethodKafkaListenerEndpoint<?, ?> endpoint) {
@@ -324,7 +323,7 @@ class RetryTopicConfigurerTests {
Object resolvedBean = handlerMethod.resolveBean(this.beanFactory);
// then
assertEquals(noOps, resolvedBean);
assertThat(resolvedBean).isEqualTo(noOps);
}
@@ -344,7 +343,7 @@ class RetryTopicConfigurerTests {
// then
then(defaultListableBeanFactory).should()
.registerBeanDefinition(eq(beanName), any(RootBeanDefinition.class));
assertTrue(NoOpsClass.class.isAssignableFrom(resolvedBean.getClass()));
assertThat(NoOpsClass.class.isAssignableFrom(resolvedBean.getClass())).isTrue();
}

10
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConstantsTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.Test;
@@ -37,9 +37,9 @@ class RetryTopicConstantsTests {
@Test
public void assertRetryTopicConstants() {
new RetryTopicConstants() { }; // for coverage
assertEquals(DEFAULT_DLT_SUFFIX, RetryTopicConstants.DEFAULT_DLT_SUFFIX);
assertEquals(DEFAULT_RETRY_SUFFIX, RetryTopicConstants.DEFAULT_RETRY_SUFFIX);
assertEquals(DEFAULT_MAX_ATTEMPTS, RetryTopicConstants.DEFAULT_MAX_ATTEMPTS);
assertEquals(NOT_SET, RetryTopicConstants.NOT_SET);
assertThat(RetryTopicConstants.DEFAULT_DLT_SUFFIX).isEqualTo(DEFAULT_DLT_SUFFIX);
assertThat(RetryTopicConstants.DEFAULT_RETRY_SUFFIX).isEqualTo(DEFAULT_RETRY_SUFFIX);
assertThat(RetryTopicConstants.DEFAULT_MAX_ATTEMPTS).isEqualTo(DEFAULT_MAX_ATTEMPTS);
assertThat(RetryTopicConstants.NOT_SET).isEqualTo(NOT_SET);
}
}

8
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicHeadersTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.Test;
@@ -35,8 +35,8 @@ class RetryTopicHeadersTests {
@Test
public void assertRetryTopicHeadersConstants() {
new RetryTopicHeaders() { }; // for coverage
assertEquals(DEFAULT_HEADER_BACKOFF_TIMESTAMP, RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);
assertEquals(DEFAULT_HEADER_ATTEMPTS, RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
assertEquals(DEFAULT_HEADER_ORIGINAL_TIMESTAMP, RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP);
assertThat(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP).isEqualTo(DEFAULT_HEADER_BACKOFF_TIMESTAMP);
assertThat(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS).isEqualTo(DEFAULT_HEADER_ATTEMPTS);
assertThat(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP).isEqualTo(DEFAULT_HEADER_ORIGINAL_TIMESTAMP);
}
}

20
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNamesTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.Test;
@@ -47,14 +47,14 @@ class RetryTopicInternalBeanNamesTests {
@Test
public void assertRetryTopicInternalBeanNamesConstants() {
new RetryTopicInternalBeanNames() { }; // for coverage
assertEquals(DESTINATION_TOPIC_PROCESSOR_NAME, RetryTopicInternalBeanNames.DESTINATION_TOPIC_PROCESSOR_NAME);
assertEquals(KAFKA_CONSUMER_BACKOFF_MANAGER, RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER);
assertEquals(RETRY_TOPIC_CONFIGURER, RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER);
assertEquals(LISTENER_CONTAINER_FACTORY_RESOLVER_NAME, RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_RESOLVER_NAME);
assertEquals(LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME, RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME);
assertEquals(DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME, RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME);
assertEquals(DESTINATION_TOPIC_CONTAINER_NAME, RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME);
assertEquals(DEFAULT_LISTENER_FACTORY_BEAN_NAME, RetryTopicInternalBeanNames.DEFAULT_LISTENER_FACTORY_BEAN_NAME);
assertEquals(DEFAULT_KAFKA_TEMPLATE_BEAN_NAME, RetryTopicInternalBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME);
assertThat(RetryTopicInternalBeanNames.DESTINATION_TOPIC_PROCESSOR_NAME).isEqualTo(DESTINATION_TOPIC_PROCESSOR_NAME);
assertThat(RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER).isEqualTo(KAFKA_CONSUMER_BACKOFF_MANAGER);
assertThat(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER).isEqualTo(RETRY_TOPIC_CONFIGURER);
assertThat(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_RESOLVER_NAME).isEqualTo(LISTENER_CONTAINER_FACTORY_RESOLVER_NAME);
assertThat(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME).isEqualTo(LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME);
assertThat(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME).isEqualTo(DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME);
assertThat(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME).isEqualTo(DESTINATION_TOPIC_CONTAINER_NAME);
assertThat(RetryTopicInternalBeanNames.DEFAULT_LISTENER_FACTORY_BEAN_NAME).isEqualTo(DEFAULT_LISTENER_FACTORY_BEAN_NAME);
assertThat(RetryTopicInternalBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME).isEqualTo(DEFAULT_KAFKA_TEMPLATE_BEAN_NAME);
}
}

56
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessorTests.java
View File

@@ -16,10 +16,8 @@
package org.springframework.kafka.retrytopic;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.BDDMockito.given;
import java.lang.reflect.Method;
@@ -107,10 +105,10 @@ class RetryableTopicAnnotationProcessorTests {
// then
RetryTopicConfigurer.EndpointHandlerMethod dltHandlerMethod = configuration.getDltHandlerMethod();
Method method = (Method) ReflectionTestUtils.getField(dltHandlerMethod, "method");
assertEquals("handleDlt", method.getName());
assertThat(method.getName()).isEqualTo("handleDlt");
assertFalse(new DestinationTopic("",
configuration.getDestinationTopicProperties().get(0)).isAlwaysRetryOnDltFailure());
assertThat(new DestinationTopic("",
configuration.getDestinationTopicProperties().get(0)).isAlwaysRetryOnDltFailure()).isFalse();
}
@Test
@@ -126,11 +124,11 @@ class RetryableTopicAnnotationProcessorTests {
// then
RetryTopicConfigurer.EndpointHandlerMethod dltHandlerMethod = configuration.getDltHandlerMethod();
Method method = (Method) ReflectionTestUtils.getField(dltHandlerMethod, "method");
assertEquals(RetryTopicConfigurer.LoggingDltListenerHandlerMethod.DEFAULT_DLT_METHOD_NAME,
method.getName());
assertThat(method.getName())
.isEqualTo(RetryTopicConfigurer.LoggingDltListenerHandlerMethod.DEFAULT_DLT_METHOD_NAME);
assertTrue(new DestinationTopic("",
configuration.getDestinationTopicProperties().get(0)).isAlwaysRetryOnDltFailure());
assertThat(new DestinationTopic("",
configuration.getDestinationTopicProperties().get(0)).isAlwaysRetryOnDltFailure()).isTrue();
}
@Test
@@ -141,8 +139,8 @@ class RetryableTopicAnnotationProcessorTests {
RetryableTopicAnnotationProcessor processor = new RetryableTopicAnnotationProcessor(beanFactory);
// given - then
assertThrows(BeanInitializationException.class, () ->
processor.processAnnotation(topics, listenWithRetry, annotation, bean));
assertThatExceptionOfType(BeanInitializationException.class)
.isThrownBy(() -> processor.processAnnotation(topics, listenWithRetry, annotation, bean));
}
@Test
@@ -158,7 +156,7 @@ class RetryableTopicAnnotationProcessorTests {
RetryableTopicAnnotationProcessor processor = new RetryableTopicAnnotationProcessor(beanFactory);
// given - then
assertThrows(BeanInitializationException.class, () ->
assertThatExceptionOfType(BeanInitializationException.class).isThrownBy(() ->
processor.processAnnotation(topics, listenWithRetryAndDlt, annotationWithDlt, beanWithDlt));
}
@@ -177,7 +175,7 @@ class RetryableTopicAnnotationProcessorTests {
bean);
DestinationTopic.Properties properties = configuration.getDestinationTopicProperties().get(0);
DestinationTopic destinationTopic = new DestinationTopic("", properties);
assertEquals(kafkaOperationsFromDefaultName, destinationTopic.getKafkaOperations());
assertThat(destinationTopic.getKafkaOperations()).isEqualTo(kafkaOperationsFromDefaultName);
}
@Test
@@ -193,7 +191,7 @@ class RetryableTopicAnnotationProcessorTests {
.processAnnotation(topics, listenWithRetry, annotation, bean);
DestinationTopic.Properties properties = configuration.getDestinationTopicProperties().get(0);
DestinationTopic destinationTopic = new DestinationTopic("", properties);
assertEquals(kafkaOperationsFromTemplateName, destinationTopic.getKafkaOperations());
assertThat(destinationTopic.getKafkaOperations()).isEqualTo(kafkaOperationsFromTemplateName);
}
@Test
@@ -211,7 +209,7 @@ class RetryableTopicAnnotationProcessorTests {
// then
DestinationTopic.Properties properties = configuration.getDestinationTopicProperties().get(0);
DestinationTopic destinationTopic = new DestinationTopic("", properties);
assertEquals(kafkaOperationsFromDefaultName, destinationTopic.getKafkaOperations());
assertThat(destinationTopic.getKafkaOperations()).isEqualTo(kafkaOperationsFromDefaultName);
}
@Test
@@ -229,13 +227,13 @@ class RetryableTopicAnnotationProcessorTests {
// then
List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
DestinationTopic destinationTopic = new DestinationTopic("", destinationTopicProperties.get(0));
assertEquals(0, destinationTopic.getDestinationDelay());
assertThat(destinationTopic.getDestinationDelay()).isEqualTo(0);
DestinationTopic destinationTopic2 = new DestinationTopic("", destinationTopicProperties.get(1));
assertEquals(1000, destinationTopic2.getDestinationDelay());
assertThat(destinationTopic2.getDestinationDelay()).isEqualTo(1000);
DestinationTopic destinationTopic3 = new DestinationTopic("", destinationTopicProperties.get(2));
assertEquals(2000, destinationTopic3.getDestinationDelay());
assertThat(destinationTopic3.getDestinationDelay()).isEqualTo(2000);
DestinationTopic destinationTopic4 = new DestinationTopic("", destinationTopicProperties.get(3));
assertEquals(0, destinationTopic4.getDestinationDelay());
assertThat(destinationTopic4.getDestinationDelay()).isEqualTo(0);
}
@@ -254,13 +252,13 @@ class RetryableTopicAnnotationProcessorTests {
// then
List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
DestinationTopic destinationTopic = new DestinationTopic("", destinationTopicProperties.get(0));
assertEquals(0, destinationTopic.getDestinationDelay());
assertThat(destinationTopic.getDestinationDelay()).isEqualTo(0);
DestinationTopic destinationTopic2 = new DestinationTopic("", destinationTopicProperties.get(1));
assertEquals(1000, destinationTopic2.getDestinationDelay());
assertThat(destinationTopic2.getDestinationDelay()).isEqualTo(1000);
DestinationTopic destinationTopic3 = new DestinationTopic("", destinationTopicProperties.get(2));
assertEquals(2000, destinationTopic3.getDestinationDelay());
assertThat(destinationTopic3.getDestinationDelay()).isEqualTo(2000);
DestinationTopic destinationTopic4 = new DestinationTopic("", destinationTopicProperties.get(3));
assertEquals(0, destinationTopic4.getDestinationDelay());
assertThat(destinationTopic4.getDestinationDelay()).isEqualTo(0);
}
@@ -279,13 +277,13 @@ class RetryableTopicAnnotationProcessorTests {
// then
List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
DestinationTopic destinationTopic = new DestinationTopic("", destinationTopicProperties.get(0));
assertEquals(0, destinationTopic.getDestinationDelay());
assertThat(destinationTopic.getDestinationDelay()).isEqualTo(0);
DestinationTopic destinationTopic2 = new DestinationTopic("", destinationTopicProperties.get(1));
assertEquals(1000, destinationTopic2.getDestinationDelay());
assertThat(destinationTopic2.getDestinationDelay()).isEqualTo(1000);
DestinationTopic destinationTopic3 = new DestinationTopic("", destinationTopicProperties.get(2));
assertEquals(1000, destinationTopic3.getDestinationDelay());
assertThat(destinationTopic3.getDestinationDelay()).isEqualTo(1000);
DestinationTopic destinationTopic4 = new DestinationTopic("", destinationTopicProperties.get(3));
assertEquals(0, destinationTopic4.getDestinationDelay());
assertThat(destinationTopic4.getDestinationDelay()).isEqualTo(0);
}

20
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicIntegrationTests.java
View File

@@ -16,8 +16,8 @@
package org.springframework.kafka.retrytopic;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Clock;
import java.util.Arrays;
@@ -95,39 +95,39 @@ public class RetryableTopicIntegrationTests {
void shouldRetryFirstTopic() {
logger.debug("Sending message to topic " + FIRST_TOPIC);
kafkaTemplate.send(FIRST_TOPIC, "Testing topic 1");
assertTrue(awaitLatch(latchContainer.countDownLatch1));
assertTrue(awaitLatch(latchContainer.customDltCountdownLatch));
assertThat(awaitLatch(latchContainer.countDownLatch1)).isTrue();
assertThat(awaitLatch(latchContainer.customDltCountdownLatch)).isTrue();
}
@Test
void shouldRetrySecondTopic() {
logger.debug("Sending message to topic " + SECOND_TOPIC);
kafkaTemplate.send(SECOND_TOPIC, "Testing topic 2");
assertTrue(awaitLatch(latchContainer.countDownLatch2));
assertTrue(awaitLatch(latchContainer.customDltCountdownLatch));
assertThat(awaitLatch(latchContainer.countDownLatch2)).isTrue();
assertThat(awaitLatch(latchContainer.customDltCountdownLatch)).isTrue();
}
@Test
void shouldRetryThirdTopicWithTimeout() {
logger.debug("Sending message to topic " + THIRD_TOPIC);
kafkaTemplate.send(THIRD_TOPIC, "Testing topic 3");
assertTrue(awaitLatch(latchContainer.countDownLatch3));
assertTrue(awaitLatch(latchContainer.countDownLatchDltOne));
assertThat(awaitLatch(latchContainer.countDownLatch3)).isTrue();
assertThat(awaitLatch(latchContainer.countDownLatchDltOne)).isTrue();
}
@Test
void shouldRetryFourthTopicWithNoDlt() {
logger.debug("Sending message to topic " + FOURTH_TOPIC);
kafkaTemplate.send(FOURTH_TOPIC, "Testing topic 4");
assertTrue(awaitLatch(latchContainer.countDownLatch4));
assertThat(awaitLatch(latchContainer.countDownLatch4)).isTrue();
}
@Test
public void shouldGoStraightToDlt() {
logger.debug("Sending message to topic " + NOT_RETRYABLE_EXCEPTION_TOPIC);
kafkaTemplate.send(NOT_RETRYABLE_EXCEPTION_TOPIC, "Testing topic with annotation 1");
assertTrue(awaitLatch(latchContainer.countDownLatchNoRetry));
assertTrue(awaitLatch(latchContainer.countDownLatchDltTwo));
assertThat(awaitLatch(latchContainer.countDownLatchNoRetry)).isTrue();
assertThat(awaitLatch(latchContainer.countDownLatchDltTwo)).isTrue();
}
private boolean awaitLatch(CountDownLatch latch) {

23
spring-kafka/src/test/java/org/springframework/kafka/support/KafkaStreamBrancherTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019 the original author or authors.
* Copyright 2019-2021 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.
@@ -16,8 +16,7 @@
package org.springframework.kafka.support;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given;
@@ -48,25 +47,25 @@ class KafkaStreamBrancherTests {
given(input.branch(eq(p1), eq(p2), any()))
.willReturn(result);
AtomicInteger invocations = new AtomicInteger(0);
assertSame(input, new KafkaStreamBrancher()
assertThat(new KafkaStreamBrancher()
.branch(
p1,
ks -> {
assertSame(result[0], ks);
assertEquals(0, invocations.getAndIncrement());
assertThat(ks).isSameAs(result[0]);
assertThat(invocations.getAndIncrement()).isEqualTo(0);
})
.defaultBranch(ks -> {
assertSame(result[2], ks);
assertEquals(2, invocations.getAndIncrement());
assertThat(ks).isSameAs(result[2]);
assertThat(invocations.getAndIncrement()).isEqualTo(2);
})
.branch(p2,
ks -> {
assertSame(result[1], ks);
assertEquals(1, invocations.getAndIncrement());
assertThat(ks).isSameAs(result[1]);
assertThat(invocations.getAndIncrement()).isEqualTo(1);
})
.onTopOf(input));
.onTopOf(input)).isSameAs(input);
assertEquals(3, invocations.get());
assertThat(invocations.get()).isEqualTo(3);
}
}