From 4f8a367c2a570c67a2141931dfdf93d88ccf29cf Mon Sep 17 00:00:00 2001 From: chickenchickenlove Date: Fri, 18 Oct 2024 23:12:53 +0900 Subject: [PATCH 1/5] Fix flaky tests in async retry by using a latch instead of sleep. --- ...pletableFutureRetryTopicScenarioTests.java | 546 ++++++++++++----- .../AsyncMonoRetryTopicScenarioTests.java | 579 ++++++++++++------ 2 files changed, 774 insertions(+), 351 deletions(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java index 6f5c321931..da0a6a0dab 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java @@ -23,9 +23,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.Random; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -107,7 +105,7 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Test void allFailCaseTest( - @Autowired TestTopicListener0 zeroTopicListener, + @Autowired TestTopicListener0 testTopicListener, @Autowired MyCustomDltProcessor myCustomDltProcessor0) { // All Fail case. String shortFailedMsg1 = "0"; @@ -126,12 +124,6 @@ void allFailCaseTest( shortFailedMsg1, shortFailedMsg2, shortFailedMsg3, - shortFailedMsg1, - shortFailedMsg2, - shortFailedMsg3, - shortFailedMsg1, - shortFailedMsg2, - shortFailedMsg3 }; String[] expectedReceivedTopics = { TEST_TOPIC0, @@ -143,12 +135,6 @@ void allFailCaseTest( expectedRetryTopic, expectedRetryTopic, expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic }; String[] expectedDltMsgs = { shortFailedMsg1, @@ -167,8 +153,8 @@ void allFailCaseTest( assertThat(destinationTopic.getDestinationName()).isEqualTo(TEST_TOPIC0 + "-retry"); - assertThat(zeroTopicListener.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); - assertThat(zeroTopicListener.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); + assertThat(testTopicListener.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); assertThat(myCustomDltProcessor0.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @@ -177,9 +163,18 @@ void allFailCaseTest( void firstShortFailAndLastLongSuccessRetryTest( @Autowired TestTopicListener1 testTopicListener1, @Autowired MyCustomDltProcessor myCustomDltProcessor1) { + // Scenario. + // 1. Short Fail msg (offset 0) + // 2. Long success msg (offset 1) -> -ing (latch wait) + // 3. Short fail msg (Retry1 offset 0) -> (latch down) + // 4. Long success msg (offset 1) -> Success! + // 5. Short fail msg (Retry2 offset 0) + // 6. Short fail msg (Retry3 offset 0) + // 7. Short fail msg (Retry4 offset 0) + // Given - String longSuccessMsg = "3"; - String shortFailedMsg = "1"; + String longSuccessMsg = testTopicListener1.LONG_SUCCESS_MSG; + String shortFailedMsg = testTopicListener1.SHORT_FAIL_MSG; DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("1-topicId", TEST_TOPIC1); String expectedRetryTopic = TEST_TOPIC1 + "-retry"; @@ -205,7 +200,6 @@ void firstShortFailAndLastLongSuccessRetryTest( shortFailedMsg }; - // When kafkaTemplate.send(TEST_TOPIC1, shortFailedMsg); kafkaTemplate.send(TEST_TOPIC1, longSuccessMsg); @@ -217,17 +211,28 @@ void firstShortFailAndLastLongSuccessRetryTest( assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); assertThat(testTopicListener1.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); assertThat(testTopicListener1.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener1.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor1.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @Test void firstLongSuccessAndLastShortFailed( - @Autowired TestTopicListener2 zero2TopicListener, + @Autowired TestTopicListener2 testTopicListener2, @Autowired MyCustomDltProcessor myCustomDltProcessor2) { + // Scenario. + // 1. Long success msg (offset 0) -> going on... (latch await) + // 2. Short fail msg (offset 1) -> done. + // 3. Short fail msg (Retry1 offset 1) -> done (latch down) + // 4. Long success msg (offset 0) -> succeed. + // 5. Short fail msg (Retry2 offset 1) + // 6. Short fail msg (Retry3 offset 1) + // 7. Short fail msg (Retry4 offset 1) + // 8. Short fail msg (dlt offset 1) + // Given - String shortFailedMsg = "1"; - String longSuccessMsg = "3"; + String shortFailedMsg = testTopicListener2.SHORT_FAIL_MSG; + String longSuccessMsg = testTopicListener2.LONG_SUCCESS_MSG; DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("2-topicId", TEST_TOPIC2); String expectedRetryTopic = TEST_TOPIC2 + "-retry"; @@ -263,34 +268,59 @@ void firstLongSuccessAndLastShortFailed( assertThat(awaitLatch(latchContainer.dltCountdownLatch2)).isTrue(); assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); - assertThat(zero2TopicListener.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); - assertThat(zero2TopicListener.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener2.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); + assertThat(testTopicListener2.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener2.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor2.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @Test - void longFailMsgTwiceThenShortSucessMsgThird( + void longFailMsgTwiceThenShortSuccessMsgThird( @Autowired TestTopicListener3 testTopicListener3, @Autowired MyCustomDltProcessor myCustomDltProcessor3) { + // Scenario + // 1. Long fail msg arrived (offset 0) -> -ing (wait latch offset 4) + // 2. Long fail msg arrived (offset 1) -> -ing (wait latch offset 1) + // 3. Short success msg arrived (offset 2) -> done + // 4. Short success msg arrived (offset 3) -> done + // 5. Short success msg arrived (offset 4) -> done (latch offset 4 count down) + // 6. Long fail msg throws error (offset 0) -> done + // 7. Long fail msg throws error (offset 1) -> done + // 8. Long fail msg (retry 1 with offset 0) -> done + // 9. Long fail msg (retry 1 with offset 1) -> done + // 10. Long fail msg (retry 2 with offset 0) -> done + // 11. Long fail msg (retry 2 with offset 1) -> done + // 12. Long fail msg (retry 3 with offset 0) -> done + // 13. Long fail msg (retry 3 with offset 1) -> done + // 14. Long fail msg (retry 4 with offset 0) -> done + // 15. Long fail msg (retry 4 with offset 1) -> done + // Given DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("3-topicId", TEST_TOPIC3); + String firstMsg = TestTopicListener3.FAIL_PREFIX + "0"; + String secondMsg = TestTopicListener3.FAIL_PREFIX + "1"; + String thirdMsg = TestTopicListener3.SUCCESS_PREFIX + "2"; + String fourthMsg = TestTopicListener3.SUCCESS_PREFIX + "3"; + String fifthMsg = TestTopicListener3.SUCCESS_PREFIX + "4"; + String expectedRetryTopic = TEST_TOPIC3 + "-retry"; + String[] expectedReceivedMsgs = { - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.SHORT_SUCCESS_MSG, - TestTopicListener3.SHORT_SUCCESS_MSG, - TestTopicListener3.SHORT_SUCCESS_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, + firstMsg, + secondMsg, + thirdMsg, + fourthMsg, + fifthMsg, + firstMsg, + secondMsg, + firstMsg, + secondMsg, + firstMsg, + secondMsg, + firstMsg, + secondMsg, }; String[] expectedReceivedTopics = { @@ -310,16 +340,16 @@ void longFailMsgTwiceThenShortSucessMsgThird( }; String[] expectedDltMsgs = { - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, + firstMsg, + secondMsg, }; // When - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.LONG_FAIL_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.LONG_FAIL_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.SHORT_SUCCESS_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.SHORT_SUCCESS_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.SHORT_SUCCESS_MSG); + kafkaTemplate.send(TEST_TOPIC3, firstMsg); + kafkaTemplate.send(TEST_TOPIC3, secondMsg); + kafkaTemplate.send(TEST_TOPIC3, thirdMsg); + kafkaTemplate.send(TEST_TOPIC3, fourthMsg); + kafkaTemplate.send(TEST_TOPIC3, fifthMsg); // Then assertThat(awaitLatch(latchContainer.countDownLatch3)).isTrue(); @@ -328,6 +358,7 @@ void longFailMsgTwiceThenShortSucessMsgThird( assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); assertThat(testTopicListener3.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); assertThat(testTopicListener3.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener3.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor3.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @@ -336,6 +367,22 @@ void longFailMsgTwiceThenShortSucessMsgThird( void longSuccessMsgTwiceThenShortFailMsgTwice( @Autowired TestTopicListener4 topicListener4, @Autowired MyCustomDltProcessor myCustomDltProcessor4) { + // Scenario + // 1. Msg arrived (offset 0) -> -ing + // 2. Msg arrived (offset 1) -> -ing + // 3. Msg arrived (offset 2) throws error -> done + // 4. Msg arrived (offset 3) throws error -> done + // 5. Msg arrived (offset 0) succeed -> done + // 6. Msg arrived (offset 1) succeed -> done + // 7. Msg arrived (retry 1, offset 2) -> done + // 8. Msg arrived (retry 1, offset 3) -> done + // 9. Msg arrived (retry 2, offset 2) -> done + // 10. Msg arrived (retry 2, offset 3) -> done + // 11. Msg arrived (retry 3, offset 2) -> done + // 12. Msg arrived (retry 3, offset 3) -> done + // 13. Msg arrived (retry 4, offset 2) -> done + // 14. Msg arrived (retry 4, offset 3) -> done + // Given DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("4-TopicId", TEST_TOPIC4); @@ -388,6 +435,7 @@ void longSuccessMsgTwiceThenShortFailMsgTwice( assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); assertThat(topicListener4.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); assertThat(topicListener4.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(topicListener4.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor4.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @@ -396,37 +444,49 @@ void longSuccessMsgTwiceThenShortFailMsgTwice( void oneLongSuccessMsgBetween100ShortFailMsg( @Autowired TestTopicListener5 topicListener5, @Autowired MyCustomDltProcessor myCustomDltProcessor5) { + // Scenario. + // 1. msgs received (offsets 0 ~ 24) -> failed. + // 2. msgs received (offset 25) -> -ing + // 3. msgs received (offset 26 ~ 49) -> failed. + // 4. msgs succeed (offset 50) -> done + // 5. msgs received (Retry1 offset 0 ~ 49 except 25) -> failed. + // 6. msgs received (Retry2 offset 0 ~ 49 except 25) -> failed. + // 7. msgs received (Retry3 offset 0 ~ 49 except 25) -> failed. + // 8. msgs received (Retry4 offset 0 ~ 49 except 25) -> failed. + // Given DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("5-TopicId", TEST_TOPIC5); String expectedRetryTopic = TEST_TOPIC5 + "-retry"; - String[] expectedReceivedMsgs = new String[501]; - for (int i = 0; i < 500; i++) { + String[] expectedReceivedMsgs = new String[148]; + for (int i = 0; i < 147; i++) { expectedReceivedMsgs[i] = TestTopicListener5.SHORT_FAIL_MSG; } - expectedReceivedMsgs[500] = TestTopicListener5.LONG_SUCCESS_MSG; + expectedReceivedMsgs[147] = TestTopicListener5.LONG_SUCCESS_MSG; - String[] expectedReceivedTopics = new String[501]; - for (int i = 0; i < 100; i++) { + String[] expectedReceivedTopics = new String[148]; + for (int i = 0; i < 49; i++) { expectedReceivedTopics[i] = TEST_TOPIC5; } - for (int i = 100; i < 500; i++) { + for (int i = 49; i < 147; i++) { expectedReceivedTopics[i] = expectedRetryTopic; } - expectedReceivedTopics[500] = TEST_TOPIC5; + expectedReceivedTopics[147] = TEST_TOPIC5; - String[] expectedDltMsgs = new String[100]; - for (int i = 0; i < 100; i++) { + String[] expectedDltMsgs = new String[49]; + for (int i = 0; i < 49; i++) { expectedDltMsgs[i] = TestTopicListener5.SHORT_FAIL_MSG; } // When - for (int i = 0; i < 100; i++) { - kafkaTemplate.send(TEST_TOPIC5, TestTopicListener5.SHORT_FAIL_MSG); - if (i == 50) { + for (int i = 0; i < 50; i++) { + if (i != 25) { + kafkaTemplate.send(TEST_TOPIC5, TestTopicListener5.SHORT_FAIL_MSG); + } + else { kafkaTemplate.send(TEST_TOPIC5, TestTopicListener5.LONG_SUCCESS_MSG); } } @@ -443,76 +503,80 @@ void oneLongSuccessMsgBetween100ShortFailMsg( } @Test - void halfSuccessMsgAndHalfFailedMsgWithRandomSleepTime( + void moreComplexAsyncScenarioTest( @Autowired TestTopicListener6 topicListener6, @Autowired @Qualifier("myCustomDltProcessor6") MyCustomDltProcessor myCustomDltProcessor6) { + // Scenario. + // 1. Fail Msg (offset 0) -> -ing + // 2. Success Msg (offset 1) -> -ing + // 3. Success Msg (offset 2) -> -ing + // 4. Fail Msg (offset 3) -> done + // 5. Success Msg (offset 4) -> -ing + // 6. Success msg succeed (offset 2) - done + // 7. Success msg succeed (offset 4) -> done + // 8. Fail Msg (Retry1 offset 3) -> done + // 9. Fail Msg (Retry2 offset 3) -> done + // 10. Success msg succeed (offset 1) -> done + // 11. Fail Msg (offset 0) -> done + // 12. Fail Msg (Retry 1 offset 0) -> done + // 13. Fail Msg (Retry 2 offset 0) -> done // Given - DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("6-TopicId", TEST_TOPIC6); + String firstMsg = TestTopicListener6.FAIL_PREFIX + "0"; + String secondMsg = TestTopicListener6.SUCCESS_PREFIX + "1"; + String thirdMsg = TestTopicListener6.SUCCESS_PREFIX + "2"; + String fourthMsg = TestTopicListener6.FAIL_PREFIX + "3"; + String fifthMsg = TestTopicListener6.SUCCESS_PREFIX + "4"; + DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("6-TopicId", TEST_TOPIC6); String expectedRetryTopic = TEST_TOPIC6 + "-retry"; - Random random = new Random(); - ConcurrentLinkedQueue q = new ConcurrentLinkedQueue<>(); - - for (int i = 0; i < 50; i++) { - int randomSleepAWhile = random.nextInt(1, 100); - String msg = randomSleepAWhile + TestTopicListener6.SUCCESS_SUFFIX; - q.add(msg); - } - - for (int i = 0; i < 50; i++) { - int randomSleepAWhile = random.nextInt(1, 100); - String msg = randomSleepAWhile + TestTopicListener6.FAIL_SUFFIX; - q.add(msg); - } - - int expectedSuccessMsgCount = 50; - int expectedFailedMsgCount = 250; + String[] expectedReceivedMsgs = { + firstMsg, + secondMsg, + thirdMsg, + fourthMsg, + fifthMsg, + fourthMsg, + fourthMsg, + firstMsg, + firstMsg + }; - int expectedReceivedOriginalTopicCount = 100; - int expectedReceivedRetryTopicCount = 200; - int expectedReceivedDltMsgCount = 50; + String[] expectedReceivedTopics = { + TEST_TOPIC6, + TEST_TOPIC6, + TEST_TOPIC6, + TEST_TOPIC6, + TEST_TOPIC6, + expectedRetryTopic, + expectedRetryTopic, + expectedRetryTopic, + expectedRetryTopic + }; + String[] expectedDltMsgs = { + TestTopicListener6.FAIL_PREFIX + "3", + TestTopicListener6.FAIL_PREFIX + "0" + }; // When - while (!q.isEmpty()) { - String successOrFailMsg = q.poll(); - kafkaTemplate.send(TEST_TOPIC6, successOrFailMsg); - } + kafkaTemplate.send(TEST_TOPIC6, firstMsg); + kafkaTemplate.send(TEST_TOPIC6, secondMsg); + kafkaTemplate.send(TEST_TOPIC6, thirdMsg); + kafkaTemplate.send(TEST_TOPIC6, fourthMsg); + kafkaTemplate.send(TEST_TOPIC6, fifthMsg); // Then assertThat(awaitLatch(latchContainer.countDownLatch6)).isTrue(); assertThat(awaitLatch(latchContainer.dltCountdownLatch6)).isTrue(); assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); + assertThat(topicListener6.receivedMsgs).containsExactly(expectedReceivedMsgs); + assertThat(topicListener6.receivedTopics).containsExactly(expectedReceivedTopics); + assertThat(topicListener6.latchWaitFailCount).isEqualTo(0); - long actualReceivedSuccessMsgCount = topicListener6.receivedMsgs.stream() - .map(s -> s.split(",")[1]) - .filter(m -> (',' + m).equals(TestTopicListener6.SUCCESS_SUFFIX)) - .count(); - - long actualReceivedFailedMsgCount = topicListener6.receivedMsgs.stream() - .map(s -> s.split(",")[1]) - .filter(m -> (',' + m).equals( - TestTopicListener6.FAIL_SUFFIX)) - .count(); - - - long actualReceivedOriginalTopicMsgCount = topicListener6.receivedTopics.stream() - .filter(topic -> topic.equals(TEST_TOPIC6)) - .count(); - - long actualReceivedRetryTopicMsgCount = topicListener6.receivedTopics.stream() - .filter(topic -> topic.equals(expectedRetryTopic)) - .count(); - - assertThat(actualReceivedSuccessMsgCount).isEqualTo(expectedSuccessMsgCount); - assertThat(actualReceivedFailedMsgCount).isEqualTo(expectedFailedMsgCount); - assertThat(actualReceivedOriginalTopicMsgCount).isEqualTo(expectedReceivedOriginalTopicCount); - assertThat(actualReceivedRetryTopicMsgCount).isEqualTo(expectedReceivedRetryTopicCount); - - assertThat(myCustomDltProcessor6.receivedMsg.size()).isEqualTo(expectedReceivedDltMsgCount); + assertThat(myCustomDltProcessor6.receivedMsg).containsExactly(expectedDltMsgs); } private boolean awaitLatch(CountDownLatch latch) { @@ -553,7 +617,7 @@ public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEI finally { container.countDownLatch0.countDown(); } - }, CompletableFuture.delayedExecutor(Integer.parseInt(message), TimeUnit.MILLISECONDS)); + }); } } @@ -574,22 +638,43 @@ static class TestTopicListener1 { private final List receivedTopics = new ArrayList<>(); + private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); + + protected final String LONG_SUCCESS_MSG = "success"; + + protected final String SHORT_FAIL_MSG = "fail"; + + protected int latchWaitFailCount = 0; + @KafkaHandler - public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { - this.receivedMsgs.add(message); + public CompletableFuture listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedTopics.add(receivedTopic); + this.receivedMsgs.add(message); return CompletableFuture.supplyAsync(() -> { try { - if (message.equals("1")) { + if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + firstRetryFailMsgLatch.await(10, TimeUnit.SECONDS); + } + } + catch (InterruptedException e) { + latchWaitFailCount += 1; + throw new RuntimeException(e); } finally { + if (receivedTopic.equals(TEST_TOPIC1 + "-retry") && + offset.equals("0")) { + firstRetryFailMsgLatch.countDown(); + } container.countDownLatch1.countDown(); } - return "Task Completed"; - }, CompletableFuture.delayedExecutor(Integer.parseInt(message), TimeUnit.MILLISECONDS)); + }); } } @@ -610,24 +695,47 @@ static class TestTopicListener2 { private final List receivedTopics = new ArrayList<>(); + private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); + + protected final String LONG_SUCCESS_MSG = "success"; + + protected final String SHORT_FAIL_MSG = "fail"; + + protected int latchWaitFailCount = 0; + @KafkaHandler - public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public CompletableFuture listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return CompletableFuture.supplyAsync(() -> { try { - if (message.equals("1")) { + if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + firstRetryFailMsgLatch.await(10, TimeUnit.SECONDS); + } + } + catch (InterruptedException e) { + latchWaitFailCount += 1; + throw new RuntimeException(e); } finally { + if (receivedTopic.equals(TEST_TOPIC2 + "-retry") && + offset.equals("1")) { + firstRetryFailMsgLatch.countDown(); + } container.countDownLatch2.countDown(); } return "Task Completed"; - }, CompletableFuture.delayedExecutor(Integer.parseInt(message), TimeUnit.MILLISECONDS)); + }); } + } @KafkaListener( @@ -646,28 +754,53 @@ static class TestTopicListener3 { private final List receivedTopics = new ArrayList<>(); - public static final String LONG_FAIL_MSG = "100"; + public static final String FAIL_PREFIX = "fail"; + + public static final String SUCCESS_PREFIX = "success"; + + private CountDownLatch successLatchCount = new CountDownLatch(3); + + private CountDownLatch offset0Latch = new CountDownLatch(1); - public static final String SHORT_SUCCESS_MSG = "1"; + protected int latchWaitFailCount = 0; @KafkaHandler - public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public CompletableFuture listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC)String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return CompletableFuture.supplyAsync(() -> { try { - if (message.equals(LONG_FAIL_MSG)) { + if (message.startsWith(FAIL_PREFIX)) { + if (receivedTopic.equals(TEST_TOPIC3)) { + if (offset.equals("0")) { + successLatchCount.await(10, TimeUnit.SECONDS); + offset0Latch.countDown(); + } + if (offset.equals("1")) { + offset0Latch.await(10, TimeUnit.SECONDS); + } + } throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + successLatchCount.countDown(); + } + } + catch (InterruptedException e) { + latchWaitFailCount += 1; + throw new RuntimeException(e); } finally { container.countDownLatch3.countDown(); } - return "Task Completed"; - }, CompletableFuture.delayedExecutor(Integer.parseInt(message), TimeUnit.MILLISECONDS)); + }); } + } @KafkaListener( @@ -686,13 +819,20 @@ static class TestTopicListener4 { private final List receivedTopics = new ArrayList<>(); - public static final String LONG_SUCCESS_MSG = "100"; + public static final String LONG_SUCCESS_MSG = "success"; - public static final String SHORT_FAIL_MSG = "1"; + public static final String SHORT_FAIL_MSG = "fail"; + + private CountDownLatch failLatchCount = new CountDownLatch(2); + + private CountDownLatch offset0Latch = new CountDownLatch(1); + + protected int latchWaitFailCount = 0; @KafkaHandler public CompletableFuture listen(String message, - @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); @@ -701,13 +841,32 @@ public CompletableFuture listen(String message, if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + failLatchCount.await(10, TimeUnit.SECONDS); + if (offset.equals("1")) { + offset0Latch.await(10, TimeUnit.SECONDS); + } + } + } + catch (InterruptedException e) { + latchWaitFailCount += 1; + throw new RuntimeException(e); } finally { + if (message.equals(SHORT_FAIL_MSG) || + receivedTopic.equals(TEST_TOPIC4)) { + failLatchCount.countDown(); + } + if (offset.equals("0") && + receivedTopic.equals(TEST_TOPIC4)) { + offset0Latch.countDown(); + } container.countDownLatch4.countDown(); } return "Task Completed"; - }, CompletableFuture.delayedExecutor(Integer.parseInt(message), TimeUnit.MILLISECONDS)); + }); } + } @KafkaListener( @@ -726,12 +885,19 @@ static class TestTopicListener5 { private final List receivedTopics = new ArrayList<>(); - public static final String LONG_SUCCESS_MSG = "100"; + public static final String LONG_SUCCESS_MSG = "success"; + + public static final String SHORT_FAIL_MSG = "fail"; - public static final String SHORT_FAIL_MSG = "1"; + private CountDownLatch failLatchCount = new CountDownLatch(24 + 49); + + protected int latchWaitFailCount = 0; @KafkaHandler - public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public CompletableFuture listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); @@ -740,14 +906,33 @@ public CompletableFuture listen(String message, @Header(KafkaHeaders.REC if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + failLatchCount.await(10, TimeUnit.SECONDS); + } + } + catch (InterruptedException e) { + latchWaitFailCount += 1; + throw new RuntimeException(e); } finally { + if (message.equals(SHORT_FAIL_MSG)) { + if (receivedTopic.equals(TEST_TOPIC5) && + Integer.valueOf(offset) > 25) { + failLatchCount.countDown(); + } + else { + if (failLatchCount.getCount() > 0) { + failLatchCount.countDown(); + } + } + } container.countDownLatch5.countDown(); } return "Task Completed"; - }, CompletableFuture.delayedExecutor(Integer.parseInt(message), TimeUnit.MILLISECONDS)); + }); } + } @KafkaListener( @@ -766,37 +951,76 @@ static class TestTopicListener6 { private final List receivedTopics = new ArrayList<>(); - public static final String SUCCESS_SUFFIX = ",s"; + public static final String SUCCESS_PREFIX = "success"; + + public static final String FAIL_PREFIX = "fail"; + + protected CountDownLatch offset1CompletedLatch = new CountDownLatch(1); + + protected CountDownLatch offset2CompletedLatch = new CountDownLatch(1); - public static final String FAIL_SUFFIX = ",f"; + protected CountDownLatch offset3RetryCompletedLatch = new CountDownLatch(3); + + protected CountDownLatch offset4ReceivedLatch = new CountDownLatch(1); + + protected int latchWaitFailCount = 0; @KafkaHandler - public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public CompletableFuture listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); - String[] split = message.split(","); - String sleepAWhile = split[0]; - String failOrSuccess = split[1]; - return CompletableFuture.supplyAsync(() -> { try { - if (failOrSuccess.equals("f")) { - throw new RuntimeException("Woooops... in topic " + receivedTopic); + if (message.startsWith(FAIL_PREFIX)) { + if (offset.equals("0")) { + if (receivedTopic.equals(TEST_TOPIC6)) { + offset1CompletedLatch.await(10, TimeUnit.SECONDS); + } + } + + if (offset.equals("3")) { + offset3RetryCompletedLatch.countDown(); + } + + throw new RuntimeException("Woooops... in topic " + receivedTopic + "msg : " + message); + } + else { + if (offset.equals("1")) { + offset3RetryCompletedLatch.await(10, TimeUnit.SECONDS); + offset1CompletedLatch.countDown(); + } + + if (offset.equals("2")) { + offset4ReceivedLatch.await(10, TimeUnit.SECONDS); + offset2CompletedLatch.countDown(); + } + + if (offset.equals("4")) { + offset4ReceivedLatch.countDown(); + offset2CompletedLatch.await(10, TimeUnit.SECONDS); + } } } + catch (InterruptedException ex) { + latchWaitFailCount += 1; + throw new RuntimeException(ex); + } finally { container.countDownLatch6.countDown(); } return "Task Completed"; - }, CompletableFuture.delayedExecutor(Integer.parseInt(sleepAWhile), TimeUnit.MILLISECONDS)); + }); } } static class CountDownLatchContainer { - static int COUNT0 = 15; + static int COUNT0 = 9; static int DLT_COUNT0 = 3; @@ -836,17 +1060,17 @@ static class CountDownLatchContainer { CountDownLatch dltCountdownLatch4 = new CountDownLatch(DLT_COUNT4); - static int COUNT5 = 501; + static int COUNT5 = 24 + 73; - static int DLT_COUNT5 = 100; + static int DLT_COUNT5 = 49; CountDownLatch countDownLatch5 = new CountDownLatch(COUNT5); CountDownLatch dltCountdownLatch5 = new CountDownLatch(DLT_COUNT5); - static int COUNT6 = 250; + static int COUNT6 = 9; - static int DLT_COUNT6 = 50; + static int DLT_COUNT6 = 2; CountDownLatch countDownLatch6 = new CountDownLatch(COUNT6); @@ -882,11 +1106,12 @@ static class RetryTopicConfigurations extends RetryTopicConfigurationSupport { static RetryTopicConfiguration createRetryTopicConfiguration( KafkaTemplate template, String topicName, - String dltBeanName) { + String dltBeanName, + int maxAttempts) { return RetryTopicConfigurationBuilder .newInstance() .fixedBackOff(50) - .maxAttempts(5) + .maxAttempts(maxAttempts) .concurrency(1) .useSingleTopicForSameIntervals() .includeTopic(topicName) @@ -900,7 +1125,8 @@ RetryTopicConfiguration testRetryTopic0(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC0, - "myCustomDltProcessor0"); + "myCustomDltProcessor0", + 3); } @Bean @@ -908,7 +1134,8 @@ RetryTopicConfiguration testRetryTopic1(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC1, - "myCustomDltProcessor1"); + "myCustomDltProcessor1", + 5); } @Bean @@ -916,7 +1143,8 @@ RetryTopicConfiguration testRetryTopic2(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC2, - "myCustomDltProcessor2"); + "myCustomDltProcessor2", + 5); } @Bean @@ -924,7 +1152,8 @@ RetryTopicConfiguration testRetryTopic3(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC3, - "myCustomDltProcessor3"); + "myCustomDltProcessor3", + 5); } @Bean @@ -932,7 +1161,8 @@ RetryTopicConfiguration testRetryTopic4(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC4, - "myCustomDltProcessor4"); + "myCustomDltProcessor4", + 5); } @Bean @@ -940,7 +1170,8 @@ RetryTopicConfiguration testRetryTopic5(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC5, - "myCustomDltProcessor5"); + "myCustomDltProcessor5", + 3); } @Bean @@ -948,7 +1179,8 @@ RetryTopicConfiguration testRetryTopic6(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC6, - "myCustomDltProcessor6"); + "myCustomDltProcessor6", + 3); } @Bean diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java index ed467cf23c..57ad1904c8 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java @@ -23,8 +23,6 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.Random; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -33,6 +31,7 @@ import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafka; @@ -102,9 +101,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Test void allFailCaseTest( - @Autowired TestTopicListener0 zeroTopicListener, + @Autowired TestTopicListener0 testTopicListener, @Autowired MyCustomDltProcessor myCustomDltProcessor0) { - // Given + // All Fail case. String shortFailedMsg1 = "0"; String shortFailedMsg2 = "1"; String shortFailedMsg3 = "2"; @@ -121,13 +120,7 @@ void allFailCaseTest( shortFailedMsg1, shortFailedMsg2, shortFailedMsg3, - shortFailedMsg1, - shortFailedMsg2, - shortFailedMsg3, - shortFailedMsg1, - shortFailedMsg2, - shortFailedMsg3 - }; + }; String[] expectedReceivedTopics = { TEST_TOPIC0, TEST_TOPIC0, @@ -138,13 +131,7 @@ void allFailCaseTest( expectedRetryTopic, expectedRetryTopic, expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic, - expectedRetryTopic - }; + }; String[] expectedDltMsgs = { shortFailedMsg1, shortFailedMsg2, @@ -162,8 +149,8 @@ void allFailCaseTest( assertThat(destinationTopic.getDestinationName()).isEqualTo(TEST_TOPIC0 + "-retry"); - assertThat(zeroTopicListener.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); - assertThat(zeroTopicListener.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); + assertThat(testTopicListener.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); assertThat(myCustomDltProcessor0.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @@ -172,9 +159,18 @@ void allFailCaseTest( void firstShortFailAndLastLongSuccessRetryTest( @Autowired TestTopicListener1 testTopicListener1, @Autowired MyCustomDltProcessor myCustomDltProcessor1) { + // Scenario. + // 1. Short Fail msg (offset 0) + // 2. Long success msg (offset 1) -> -ing (latch wait) + // 3. Short fail msg (Retry1 offset 0) -> (latch down) + // 4. Long success msg (offset 1) -> Success! + // 5. Short fail msg (Retry2 offset 0) + // 6. Short fail msg (Retry3 offset 0) + // 7. Short fail msg (Retry4 offset 0) + // Given - String longSuccessMsg = "3"; - String shortFailedMsg = "1"; + String longSuccessMsg = testTopicListener1.LONG_SUCCESS_MSG; + String shortFailedMsg = testTopicListener1.SHORT_FAIL_MSG; DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("1-topicId", TEST_TOPIC1); String expectedRetryTopic = TEST_TOPIC1 + "-retry"; @@ -211,17 +207,28 @@ void firstShortFailAndLastLongSuccessRetryTest( assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); assertThat(testTopicListener1.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); assertThat(testTopicListener1.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener1.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor1.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @Test void firstLongSuccessAndLastShortFailed( - @Autowired TestTopicListener2 zero2TopicListener, + @Autowired TestTopicListener2 testTopicListener2, @Autowired MyCustomDltProcessor myCustomDltProcessor2) { + // Scenario. + // 1. Long success msg (offset 0) -> going on... (latch await) + // 2. Short fail msg (offset 1) -> done. + // 3. Short fail msg (Retry1 offset 1) -> done (latch down) + // 4. Long success msg (offset 0) -> succeed. + // 5. Short fail msg (Retry2 offset 1) + // 6. Short fail msg (Retry3 offset 1) + // 7. Short fail msg (Retry4 offset 1) + // 8. Short fail msg (dlt offset 1) + // Given - String shortFailedMsg = "1"; - String longSuccessMsg = "3"; + String shortFailedMsg = testTopicListener2.SHORT_FAIL_MSG; + String longSuccessMsg = testTopicListener2.LONG_SUCCESS_MSG; DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("2-topicId", TEST_TOPIC2); String expectedRetryTopic = TEST_TOPIC2 + "-retry"; @@ -252,12 +259,14 @@ void firstLongSuccessAndLastShortFailed( kafkaTemplate.send(TEST_TOPIC2, shortFailedMsg); // Then + assertThat(awaitLatch(latchContainer.countDownLatch2)).isTrue(); assertThat(awaitLatch(latchContainer.dltCountdownLatch2)).isTrue(); assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); - assertThat(zero2TopicListener.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); - assertThat(zero2TopicListener.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener2.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); + assertThat(testTopicListener2.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener2.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor2.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @@ -266,24 +275,48 @@ void firstLongSuccessAndLastShortFailed( void longFailMsgTwiceThenShortSuccessMsgThird( @Autowired TestTopicListener3 testTopicListener3, @Autowired MyCustomDltProcessor myCustomDltProcessor3) { + // Scenario + // 1. Long fail msg arrived (offset 0) -> -ing (wait latch offset 4) + // 2. Long fail msg arrived (offset 1) -> -ing (wait latch offset 1) + // 3. Short success msg arrived (offset 2) -> done + // 4. Short success msg arrived (offset 3) -> done + // 5. Short success msg arrived (offset 4) -> done (latch offset 4 count down) + // 6. Long fail msg throws error (offset 0) -> done + // 7. Long fail msg throws error (offset 1) -> done + // 8. Long fail msg (retry 1 with offset 0) -> done + // 9. Long fail msg (retry 1 with offset 1) -> done + // 10. Long fail msg (retry 2 with offset 0) -> done + // 11. Long fail msg (retry 2 with offset 1) -> done + // 12. Long fail msg (retry 3 with offset 0) -> done + // 13. Long fail msg (retry 3 with offset 1) -> done + // 14. Long fail msg (retry 4 with offset 0) -> done + // 15. Long fail msg (retry 4 with offset 1) -> done + // Given DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("3-topicId", TEST_TOPIC3); + String firstMsg = TestTopicListener3.FAIL_PREFIX + "0"; + String secondMsg = TestTopicListener3.FAIL_PREFIX + "1"; + String thirdMsg = TestTopicListener3.SUCCESS_PREFIX + "2"; + String fourthMsg = TestTopicListener3.SUCCESS_PREFIX + "3"; + String fifthMsg = TestTopicListener3.SUCCESS_PREFIX + "4"; + String expectedRetryTopic = TEST_TOPIC3 + "-retry"; + String[] expectedReceivedMsgs = { - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.SHORT_SUCCESS_MSG, - TestTopicListener3.SHORT_SUCCESS_MSG, - TestTopicListener3.SHORT_SUCCESS_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, + firstMsg, + secondMsg, + thirdMsg, + fourthMsg, + fifthMsg, + firstMsg, + secondMsg, + firstMsg, + secondMsg, + firstMsg, + secondMsg, + firstMsg, + secondMsg, }; String[] expectedReceivedTopics = { @@ -303,16 +336,16 @@ void longFailMsgTwiceThenShortSuccessMsgThird( }; String[] expectedDltMsgs = { - TestTopicListener3.LONG_FAIL_MSG, - TestTopicListener3.LONG_FAIL_MSG, + firstMsg, + secondMsg, }; // When - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.LONG_FAIL_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.LONG_FAIL_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.SHORT_SUCCESS_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.SHORT_SUCCESS_MSG); - kafkaTemplate.send(TEST_TOPIC3, TestTopicListener3.SHORT_SUCCESS_MSG); + kafkaTemplate.send(TEST_TOPIC3, firstMsg); + kafkaTemplate.send(TEST_TOPIC3, secondMsg); + kafkaTemplate.send(TEST_TOPIC3, thirdMsg); + kafkaTemplate.send(TEST_TOPIC3, fourthMsg); + kafkaTemplate.send(TEST_TOPIC3, fifthMsg); // Then assertThat(awaitLatch(latchContainer.countDownLatch3)).isTrue(); @@ -321,6 +354,7 @@ void longFailMsgTwiceThenShortSuccessMsgThird( assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); assertThat(testTopicListener3.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); assertThat(testTopicListener3.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(testTopicListener3.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor3.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @@ -329,6 +363,22 @@ void longFailMsgTwiceThenShortSuccessMsgThird( void longSuccessMsgTwiceThenShortFailMsgTwice( @Autowired TestTopicListener4 topicListener4, @Autowired MyCustomDltProcessor myCustomDltProcessor4) { + // Scenario + // 1. Msg arrived (offset 0) -> -ing + // 2. Msg arrived (offset 1) -> -ing + // 3. Msg arrived (offset 2) throws error -> done + // 4. Msg arrived (offset 3) throws error -> done + // 5. Msg arrived (offset 0) succeed -> done + // 6. Msg arrived (offset 1) succeed -> done + // 7. Msg arrived (retry 1, offset 2) -> done + // 8. Msg arrived (retry 1, offset 3) -> done + // 9. Msg arrived (retry 2, offset 2) -> done + // 10. Msg arrived (retry 2, offset 3) -> done + // 11. Msg arrived (retry 3, offset 2) -> done + // 12. Msg arrived (retry 3, offset 3) -> done + // 13. Msg arrived (retry 4, offset 2) -> done + // 14. Msg arrived (retry 4, offset 3) -> done + // Given DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("4-TopicId", TEST_TOPIC4); @@ -361,7 +411,7 @@ void longSuccessMsgTwiceThenShortFailMsgTwice( expectedRetryTopic, expectedRetryTopic, expectedRetryTopic, - }; + }; String[] expectedDltMsgs = { TestTopicListener4.SHORT_FAIL_MSG, @@ -381,44 +431,58 @@ void longSuccessMsgTwiceThenShortFailMsgTwice( assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); assertThat(topicListener4.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); assertThat(topicListener4.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(topicListener4.latchWaitFailCount).isEqualTo(0); assertThat(myCustomDltProcessor4.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } @Test - void oneLongSuccessMsgBetween100ShortFailMsgs( + void oneLongSuccessMsgBetween100ShortFailMsg( @Autowired TestTopicListener5 topicListener5, @Autowired MyCustomDltProcessor myCustomDltProcessor5) { + // Scenario. + // 1. msgs received (offsets 0 ~ 24) -> failed. + // 2. msgs received (offset 25) -> -ing + // 3. msgs received (offset 26 ~ 49) -> failed. + // 4. msgs succeed (offset 50) -> done + // 5. msgs received (Retry1 offset 0 ~ 49 except 25) -> failed. + // 6. msgs received (Retry2 offset 0 ~ 49 except 25) -> failed. + // 7. msgs received (Retry3 offset 0 ~ 49 except 25) -> failed. + // 8. msgs received (Retry4 offset 0 ~ 49 except 25) -> failed. + // Given DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("5-TopicId", TEST_TOPIC5); String expectedRetryTopic = TEST_TOPIC5 + "-retry"; - String[] expectedReceivedMsgs = new String[501]; - for (int i = 0; i < 500; i++) { + String[] expectedReceivedMsgs = new String[148]; + for (int i = 0; i < 147; i++) { expectedReceivedMsgs[i] = TestTopicListener5.SHORT_FAIL_MSG; } - expectedReceivedMsgs[500] = TestTopicListener5.LONG_SUCCESS_MSG; + expectedReceivedMsgs[147] = TestTopicListener5.LONG_SUCCESS_MSG; + - String[] expectedReceivedTopics = new String[501]; - for (int i = 0; i < 100; i++) { + String[] expectedReceivedTopics = new String[148]; + for (int i = 0; i < 49; i++) { expectedReceivedTopics[i] = TEST_TOPIC5; } - for (int i = 100; i < 500; i++) { + for (int i = 49; i < 147; i++) { expectedReceivedTopics[i] = expectedRetryTopic; } - expectedReceivedTopics[500] = TEST_TOPIC5; + expectedReceivedTopics[147] = TEST_TOPIC5; - String[] expectedDltMsgs = new String[100]; - for (int i = 0; i < 100; i++) { + String[] expectedDltMsgs = new String[49]; + for (int i = 0; i < 49; i++) { expectedDltMsgs[i] = TestTopicListener5.SHORT_FAIL_MSG; } // When - for (int i = 0; i < 100; i++) { - kafkaTemplate.send(TEST_TOPIC5, TestTopicListener5.SHORT_FAIL_MSG); - if (i == 50) { + for (int i = 0; i < 50; i++) { + if (i != 25) { + kafkaTemplate.send(TEST_TOPIC5, TestTopicListener5.SHORT_FAIL_MSG); + } + else { kafkaTemplate.send(TEST_TOPIC5, TestTopicListener5.LONG_SUCCESS_MSG); } } @@ -435,75 +499,81 @@ void oneLongSuccessMsgBetween100ShortFailMsgs( } @Test - void halfSuccessMsgAndHalfFailedMsgWithRandomSleepTime( + void moreComplexAsyncScenarioTest( @Autowired TestTopicListener6 topicListener6, - @Autowired MyCustomDltProcessor myCustomDltProcessor6) { + @Autowired @Qualifier("myCustomDltProcessor6") + MyCustomDltProcessor myCustomDltProcessor6) { + // Scenario. + // 1. Fail Msg (offset 0) -> -ing + // 2. Success Msg (offset 1) -> -ing + // 3. Success Msg (offset 2) -> -ing + // 4. Fail Msg (offset 3) -> done + // 5. Success Msg (offset 4) -> -ing + // 6. Success msg succeed (offset 2) - done + // 7. Success msg succeed (offset 4) -> done + // 8. Fail Msg (Retry1 offset 3) -> done + // 9. Fail Msg (Retry2 offset 3) -> done + // 10. Success msg succeed (offset 1) -> done + // 11. Fail Msg (offset 0) -> done + // 12. Fail Msg (Retry 1 offset 0) -> done + // 13. Fail Msg (Retry 2 offset 0) -> done + // Given - DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("6-TopicId", TEST_TOPIC6); + String firstMsg = TestTopicListener6.FAIL_PREFIX + "0"; + String secondMsg = TestTopicListener6.SUCCESS_PREFIX + "1"; + String thirdMsg = TestTopicListener6.SUCCESS_PREFIX + "2"; + String fourthMsg = TestTopicListener6.FAIL_PREFIX + "3"; + String fifthMsg = TestTopicListener6.SUCCESS_PREFIX + "4"; + DestinationTopic destinationTopic = topicContainer.getNextDestinationTopicFor("6-TopicId", TEST_TOPIC6); String expectedRetryTopic = TEST_TOPIC6 + "-retry"; - Random random = new Random(); - ConcurrentLinkedQueue q = new ConcurrentLinkedQueue<>(); - - for (int i = 0; i < 50; i++) { - int randomSleepAWhile = random.nextInt(1, 100); - String msg = String.valueOf(randomSleepAWhile) + TestTopicListener6.SUCCESS_SUFFIX; - q.add(msg); - } - - for (int i = 0; i < 50; i++) { - int randomSleepAWhile = random.nextInt(1, 100); - String msg = String.valueOf(randomSleepAWhile) + TestTopicListener6.FAIL_SUFFIX; - q.add(msg); - } - - int expectedSuccessMsgCount = 50; - int expectedFailedMsgCount = 250; + String[] expectedReceivedMsgs = { + firstMsg, + secondMsg, + thirdMsg, + fourthMsg, + fifthMsg, + fourthMsg, + fourthMsg, + firstMsg, + firstMsg + }; - int expectedReceivedOriginalTopicCount = 100; - int expectedReceivedRetryTopicCount = 200; - int expectedReceivedDltMsgCount = 50; + String[] expectedReceivedTopics = { + TEST_TOPIC6, + TEST_TOPIC6, + TEST_TOPIC6, + TEST_TOPIC6, + TEST_TOPIC6, + expectedRetryTopic, + expectedRetryTopic, + expectedRetryTopic, + expectedRetryTopic + }; + String[] expectedDltMsgs = { + TestTopicListener6.FAIL_PREFIX + "3", + TestTopicListener6.FAIL_PREFIX + "0" + }; // When - while (!q.isEmpty()) { - String successOrFailMsg = q.poll(); - kafkaTemplate.send(TEST_TOPIC6, successOrFailMsg); - } + kafkaTemplate.send(TEST_TOPIC6, firstMsg); + kafkaTemplate.send(TEST_TOPIC6, secondMsg); + kafkaTemplate.send(TEST_TOPIC6, thirdMsg); + kafkaTemplate.send(TEST_TOPIC6, fourthMsg); + kafkaTemplate.send(TEST_TOPIC6, fifthMsg); // Then assertThat(awaitLatch(latchContainer.countDownLatch6)).isTrue(); assertThat(awaitLatch(latchContainer.dltCountdownLatch6)).isTrue(); assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); + assertThat(topicListener6.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); + assertThat(topicListener6.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); + assertThat(topicListener6.latchWaitFailCount).isEqualTo(0); - long actualReceivedSuccessMsgCount = topicListener6.receivedMsgs.stream() - .map(s -> s.split(",")[1]) - .filter(m -> (',' + m).equals(TestTopicListener6.SUCCESS_SUFFIX)) - .count(); - - long actualReceivedFailedMsgCount = topicListener6.receivedMsgs.stream() - .map(s -> s.split(",")[1]) - .filter(m -> (',' + m).equals( - TestTopicListener6.FAIL_SUFFIX)) - .count(); - - - long actualReceivedOriginalTopicMsgCount = topicListener6.receivedTopics.stream() - .filter(topic -> topic.equals(TEST_TOPIC6)) - .count(); - - long actualReceivedRetryTopicMsgCount = topicListener6.receivedTopics.stream() - .filter(topic -> topic.equals(expectedRetryTopic)) - .count(); - - assertThat(actualReceivedSuccessMsgCount).isEqualTo(expectedSuccessMsgCount); - assertThat(actualReceivedFailedMsgCount).isEqualTo(expectedFailedMsgCount); - assertThat(actualReceivedOriginalTopicMsgCount).isEqualTo(expectedReceivedOriginalTopicCount); - assertThat(actualReceivedRetryTopicMsgCount).isEqualTo(expectedReceivedRetryTopicCount); - - assertThat(myCustomDltProcessor6.receivedMsg.size()).isEqualTo(expectedReceivedDltMsgCount); + assertThat(myCustomDltProcessor6.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } private boolean awaitLatch(CountDownLatch latch) { @@ -514,6 +584,7 @@ private boolean awaitLatch(CountDownLatch latch) { fail(e.getMessage()); throw new RuntimeException(e); } + } @KafkaListener( @@ -540,13 +611,10 @@ public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) St try { throw new RuntimeException("Woooops... in topic " + receivedTopic); } - catch (Exception e) { - throw e; - } finally { container.countDownLatch0.countDown(); } - }).then(); + }); } } @@ -567,32 +635,45 @@ static class TestTopicListener1 { private final List receivedTopics = new ArrayList<>(); + private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); + + protected final String LONG_SUCCESS_MSG = "success"; + + protected final String SHORT_FAIL_MSG = "fail"; + + protected int latchWaitFailCount = 0; + @KafkaHandler - public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { - this.receivedMsgs.add(message); + public Mono listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedTopics.add(receivedTopic); - + this.receivedMsgs.add(message); return Mono.fromCallable(() -> { try { - Thread.sleep(Integer.parseInt(message)); - if (message.equals("1")) { + if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + firstRetryFailMsgLatch.await(10, TimeUnit.SECONDS); + } } catch (InterruptedException e) { + latchWaitFailCount += 1; throw new RuntimeException(e); } - catch (RuntimeException e) { - throw e; - } finally { + if (receivedTopic.equals(TEST_TOPIC1 + "-retry") && + offset.equals("0")) { + firstRetryFailMsgLatch.countDown(); + } container.countDownLatch1.countDown(); } - return "Task Completed"; }); - } + } @KafkaListener( @@ -611,31 +692,47 @@ static class TestTopicListener2 { private final List receivedTopics = new ArrayList<>(); + private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); + + protected final String LONG_SUCCESS_MSG = "success"; + + protected final String SHORT_FAIL_MSG = "fail"; + + protected int latchWaitFailCount = 0; + @KafkaHandler - public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public Mono listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return Mono.fromCallable(() -> { try { - Thread.sleep(Integer.parseInt(message)); - if (message.equals("1")) { + if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + firstRetryFailMsgLatch.await(10, TimeUnit.SECONDS); + } } catch (InterruptedException e) { + latchWaitFailCount += 1; throw new RuntimeException(e); } - catch (RuntimeException e) { - throw e; - } finally { + if (receivedTopic.equals(TEST_TOPIC2 + "-retry") && + offset.equals("1")) { + firstRetryFailMsgLatch.countDown(); + } container.countDownLatch2.countDown(); } return "Task Completed"; }); } + } @KafkaListener( @@ -654,37 +751,53 @@ static class TestTopicListener3 { private final List receivedTopics = new ArrayList<>(); - public static final String LONG_FAIL_MSG = "100"; + public static final String FAIL_PREFIX = "fail"; + + public static final String SUCCESS_PREFIX = "success"; - public static final String SHORT_SUCCESS_MSG = "1"; + private CountDownLatch successLatchCount = new CountDownLatch(3); + + private CountDownLatch offset0Latch = new CountDownLatch(1); + + protected int latchWaitFailCount = 0; @KafkaHandler - public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public Mono listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC)String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return Mono.fromCallable(() -> { try { - Thread.sleep(Integer.parseInt(message)); - if (message.equals(LONG_FAIL_MSG)) { + if (message.startsWith(FAIL_PREFIX)) { + if (receivedTopic.equals(TEST_TOPIC3)) { + if (offset.equals("0")) { + successLatchCount.await(10, TimeUnit.SECONDS); + offset0Latch.countDown(); + } + if (offset.equals("1")) { + offset0Latch.await(10, TimeUnit.SECONDS); + } + } throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + successLatchCount.countDown(); + } } catch (InterruptedException e) { + latchWaitFailCount += 1; throw new RuntimeException(e); } - catch (RuntimeException e) { - throw e; - } finally { container.countDownLatch3.countDown(); } - return "Task Completed"; }); - - } + } @KafkaListener( @@ -703,36 +816,54 @@ static class TestTopicListener4 { private final List receivedTopics = new ArrayList<>(); - public static final String LONG_SUCCESS_MSG = "100"; + public static final String LONG_SUCCESS_MSG = "success"; + + public static final String SHORT_FAIL_MSG = "fail"; + + private CountDownLatch failLatchCount = new CountDownLatch(2); + + private CountDownLatch offset0Latch = new CountDownLatch(1); - public static final String SHORT_FAIL_MSG = "1"; + protected int latchWaitFailCount = 0; @KafkaHandler - public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public Mono listen(String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return Mono.fromCallable(() -> { try { - Thread.sleep(Integer.parseInt(message)); if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + failLatchCount.await(10, TimeUnit.SECONDS); + if (offset.equals("1")) { + offset0Latch.await(10, TimeUnit.SECONDS); + } + } } catch (InterruptedException e) { + latchWaitFailCount += 1; throw new RuntimeException(e); } - catch (RuntimeException e) { - throw e; - } finally { + if (message.equals(SHORT_FAIL_MSG) || + receivedTopic.equals(TEST_TOPIC4)) { + failLatchCount.countDown(); + } + if (offset.equals("0") && + receivedTopic.equals(TEST_TOPIC4)) { + offset0Latch.countDown(); + } container.countDownLatch4.countDown(); } - return "Task Completed"; }); - } + } @KafkaListener( @@ -751,35 +882,54 @@ static class TestTopicListener5 { private final List receivedTopics = new ArrayList<>(); - public static final String LONG_SUCCESS_MSG = "100"; + public static final String LONG_SUCCESS_MSG = "success"; + + public static final String SHORT_FAIL_MSG = "fail"; - public static final String SHORT_FAIL_MSG = "1"; + private CountDownLatch failLatchCount = new CountDownLatch(24 + 49); + + protected int latchWaitFailCount = 0; @KafkaHandler - public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public Mono listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return Mono.fromCallable(() -> { try { - Thread.sleep(Integer.parseInt(message)); if (message.equals(SHORT_FAIL_MSG)) { throw new RuntimeException("Woooops... in topic " + receivedTopic); } + else { + failLatchCount.await(10, TimeUnit.SECONDS); + } } catch (InterruptedException e) { + latchWaitFailCount += 1; throw new RuntimeException(e); } - catch (RuntimeException e) { - throw e; - } finally { + if (message.equals(SHORT_FAIL_MSG)) { + if (receivedTopic.equals(TEST_TOPIC5) && + Integer.valueOf(offset) > 25) { + failLatchCount.countDown(); + } + else { + if (failLatchCount.getCount() > 0) { + failLatchCount.countDown(); + } + } + } container.countDownLatch5.countDown(); } return "Task Completed"; }); } + } @KafkaListener( @@ -798,31 +948,63 @@ static class TestTopicListener6 { private final List receivedTopics = new ArrayList<>(); - public static final String SUCCESS_SUFFIX = ",s"; + public static final String SUCCESS_PREFIX = "success"; + + public static final String FAIL_PREFIX = "fail"; + + protected CountDownLatch offset1CompletedLatch = new CountDownLatch(1); + + protected CountDownLatch offset2CompletedLatch = new CountDownLatch(1); + + protected CountDownLatch offset3RetryCompletedLatch = new CountDownLatch(3); + + protected CountDownLatch offset4ReceivedLatch = new CountDownLatch(1); - public static final String FAIL_SUFFIX = ",f"; + protected int latchWaitFailCount = 0; @KafkaHandler - public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { + public Mono listen( + String message, + @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic, + @Header(KafkaHeaders.OFFSET) String offset) { this.receivedMsgs.add(message); this.receivedTopics.add(receivedTopic); return Mono.fromCallable(() -> { - String[] split = message.split(","); - String sleepAWhile = split[0]; - String failOrSuccess = split[1]; - try { - Thread.sleep(Integer.parseInt(sleepAWhile)); - if (failOrSuccess.equals("f")) { - throw new RuntimeException("Woooops... in topic " + receivedTopic); + if (message.startsWith(FAIL_PREFIX)) { + if (offset.equals("0")) { + if (receivedTopic.equals(TEST_TOPIC6)) { + offset1CompletedLatch.await(10, TimeUnit.SECONDS); + } + } + + if (offset.equals("3")) { + offset3RetryCompletedLatch.countDown(); + } + + throw new RuntimeException("Woooops... in topic " + receivedTopic + "msg : " + message); + } + else { + if (offset.equals("1")) { + offset3RetryCompletedLatch.await(10, TimeUnit.SECONDS); + offset1CompletedLatch.countDown(); + } + + if (offset.equals("2")) { + offset4ReceivedLatch.await(10, TimeUnit.SECONDS); + offset2CompletedLatch.countDown(); + } + + if (offset.equals("4")) { + offset4ReceivedLatch.countDown(); + offset2CompletedLatch.await(10, TimeUnit.SECONDS); + } } } - catch (InterruptedException e) { - throw new RuntimeException(e); - } - catch (RuntimeException e) { - throw e; + catch (InterruptedException ex) { + latchWaitFailCount += 1; + throw new RuntimeException(ex); } finally { container.countDownLatch6.countDown(); @@ -835,7 +1017,7 @@ public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) static class CountDownLatchContainer { - static int COUNT0 = 15; + static int COUNT0 = 9; static int DLT_COUNT0 = 3; @@ -875,17 +1057,17 @@ static class CountDownLatchContainer { CountDownLatch dltCountdownLatch4 = new CountDownLatch(DLT_COUNT4); - static int COUNT5 = 501; + static int COUNT5 = 24 + 73; - static int DLT_COUNT5 = 100; + static int DLT_COUNT5 = 49; CountDownLatch countDownLatch5 = new CountDownLatch(COUNT5); CountDownLatch dltCountdownLatch5 = new CountDownLatch(DLT_COUNT5); - static int COUNT6 = 250; + static int COUNT6 = 9; - static int DLT_COUNT6 = 50; + static int DLT_COUNT6 = 2; CountDownLatch countDownLatch6 = new CountDownLatch(COUNT6); @@ -897,8 +1079,7 @@ static class MyCustomDltProcessor { final List receivedMsg = new ArrayList<>(); - MyCustomDltProcessor(KafkaTemplate kafkaTemplate, - CountDownLatch latch) { + MyCustomDltProcessor(KafkaTemplate kafkaTemplate, CountDownLatch latch) { this.kafkaTemplate = kafkaTemplate; this.latch = latch; } @@ -918,13 +1099,15 @@ static class RetryTopicConfigurations extends RetryTopicConfigurationSupport { private static final String DLT_METHOD_NAME = "processDltMessage"; - static RetryTopicConfiguration createRetryTopicConfiguration(KafkaTemplate template, - String topicName, - String dltBeanName) { + static RetryTopicConfiguration createRetryTopicConfiguration( + KafkaTemplate template, + String topicName, + String dltBeanName, + int maxAttempts) { return RetryTopicConfigurationBuilder .newInstance() .fixedBackOff(50) - .maxAttempts(5) + .maxAttempts(maxAttempts) .concurrency(1) .useSingleTopicForSameIntervals() .includeTopic(topicName) @@ -938,7 +1121,8 @@ RetryTopicConfiguration testRetryTopic0(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC0, - "myCustomDltProcessor0"); + "myCustomDltProcessor0", + 3); } @Bean @@ -946,7 +1130,8 @@ RetryTopicConfiguration testRetryTopic1(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC1, - "myCustomDltProcessor1"); + "myCustomDltProcessor1", + 5); } @Bean @@ -954,7 +1139,8 @@ RetryTopicConfiguration testRetryTopic2(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC2, - "myCustomDltProcessor2"); + "myCustomDltProcessor2", + 5); } @Bean @@ -962,7 +1148,8 @@ RetryTopicConfiguration testRetryTopic3(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC3, - "myCustomDltProcessor3"); + "myCustomDltProcessor3", + 5); } @Bean @@ -970,7 +1157,8 @@ RetryTopicConfiguration testRetryTopic4(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC4, - "myCustomDltProcessor4"); + "myCustomDltProcessor4", + 5); } @Bean @@ -978,7 +1166,8 @@ RetryTopicConfiguration testRetryTopic5(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC5, - "myCustomDltProcessor5"); + "myCustomDltProcessor5", + 3); } @Bean @@ -986,7 +1175,8 @@ RetryTopicConfiguration testRetryTopic6(KafkaTemplate template) return createRetryTopicConfiguration( template, TEST_TOPIC6, - "myCustomDltProcessor6"); + "myCustomDltProcessor6", + 3); } @Bean @@ -1125,6 +1315,7 @@ ProducerFactory producerFactory() { KafkaTemplate kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } + } @EnableKafka From 373d385e11bf880b298c74493b716f6f462885f4 Mon Sep 17 00:00:00 2001 From: chickenchickenlove Date: Fri, 18 Oct 2024 23:33:14 +0900 Subject: [PATCH 2/5] Remove conditional test from async retry scenario tests. --- .../AsyncCompletableFutureRetryTopicScenarioTests.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java index da0a6a0dab..20200084cb 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java @@ -74,8 +74,6 @@ @DirtiesContext @EmbeddedKafka @TestPropertySource(properties = { "five.attempts=5", "kafka.template=customKafkaTemplate"}) -@DisabledIfEnvironmentVariable(named = "GITHUB_ACTION", matches = ".*?", - disabledReason = "Fails sporadically. Perhaps uses too much Apache Kafka resources") public class AsyncCompletableFutureRetryTopicScenarioTests { private final static String MAIN_TOPIC_CONTAINER_FACTORY = "kafkaListenerContainerFactory"; From 46b445278ea55b73b48081b796ce0096d04d7ab6 Mon Sep 17 00:00:00 2001 From: chickenchickenlove Date: Fri, 18 Oct 2024 23:36:30 +0900 Subject: [PATCH 3/5] Modify name of a test case --- .../AsyncCompletableFutureRetryTopicScenarioTests.java | 2 +- .../kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java index 20200084cb..e9948dae17 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java @@ -439,7 +439,7 @@ void longSuccessMsgTwiceThenShortFailMsgTwice( } @Test - void oneLongSuccessMsgBetween100ShortFailMsg( + void oneLongSuccessMsgBetween49ShortFailMsg( @Autowired TestTopicListener5 topicListener5, @Autowired MyCustomDltProcessor myCustomDltProcessor5) { // Scenario. diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java index 57ad1904c8..a1ca877b89 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java @@ -437,7 +437,7 @@ void longSuccessMsgTwiceThenShortFailMsgTwice( } @Test - void oneLongSuccessMsgBetween100ShortFailMsg( + void oneLongSuccessMsgBetween49ShortFailMsg( @Autowired TestTopicListener5 topicListener5, @Autowired MyCustomDltProcessor myCustomDltProcessor5) { // Scenario. From 676d12309ae8f9f81c7399177701cdae3241013f Mon Sep 17 00:00:00 2001 From: chickenchickenlove Date: Fri, 18 Oct 2024 23:39:27 +0900 Subject: [PATCH 4/5] Fixes lint error : remove unused import. --- .../AsyncCompletableFutureRetryTopicScenarioTests.java | 1 - 1 file changed, 1 deletion(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java index e9948dae17..59da1b207b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java @@ -30,7 +30,6 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.StringSerializer; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; From 6a3f20483d4559956f07140bb8a24878f8e7a2ab Mon Sep 17 00:00:00 2001 From: chickenchickenlove Date: Fri, 18 Oct 2024 23:54:24 +0900 Subject: [PATCH 5/5] Use containsExactlyInAnyOrder instead of containsExactly. --- .../AsyncCompletableFutureRetryTopicScenarioTests.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java index 59da1b207b..a4db2790c6 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java @@ -569,11 +569,11 @@ void moreComplexAsyncScenarioTest( assertThat(awaitLatch(latchContainer.dltCountdownLatch6)).isTrue(); assertThat(destinationTopic.getDestinationName()).isEqualTo(expectedRetryTopic); - assertThat(topicListener6.receivedMsgs).containsExactly(expectedReceivedMsgs); - assertThat(topicListener6.receivedTopics).containsExactly(expectedReceivedTopics); + assertThat(topicListener6.receivedMsgs).containsExactlyInAnyOrder(expectedReceivedMsgs); + assertThat(topicListener6.receivedTopics).containsExactlyInAnyOrder(expectedReceivedTopics); assertThat(topicListener6.latchWaitFailCount).isEqualTo(0); - assertThat(myCustomDltProcessor6.receivedMsg).containsExactly(expectedDltMsgs); + assertThat(myCustomDltProcessor6.receivedMsg).containsExactlyInAnyOrder(expectedDltMsgs); } private boolean awaitLatch(CountDownLatch latch) {