From c58e1c3b372c33cd357980b9415f30664cab1e18 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Mar 2025 15:55:10 -0500 Subject: [PATCH] Fix async tests for `Collections.synchronizedList()` The `AsyncCompletableFutureRetryTopicScenarioTests` & `AsyncMonoRetryTopicScenarioTests` use `new ArrayList()` to collect consumed messages and then `CountDownLatch` to fulfill expectations. Turns out the `CountDownLatch` might be fulfilled and respective assert in the test would pass, but the next assertion for the mentioned list might fail. The idea that state of the `ArrayList` cannot be predictable in between threads. So, suggestion is to use `Collections.synchronizedList()` instead for better memory barrier management --- ...pletableFutureRetryTopicScenarioTests.java | 32 +++++++++---------- .../AsyncMonoRetryTopicScenarioTests.java | 32 +++++++++---------- 2 files changed, 32 insertions(+), 32 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 f1b939b6..44c98622 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 @@ -1,5 +1,5 @@ /* - * Copyright 2024 the original author or authors. + * Copyright 2024-2025 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. @@ -602,9 +602,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - private final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); @KafkaHandler public CompletableFuture listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { @@ -634,9 +634,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - private final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); @@ -691,9 +691,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); @@ -750,9 +750,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String FAIL_PREFIX = "fail"; @@ -815,9 +815,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String LONG_SUCCESS_MSG = "success"; @@ -881,9 +881,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String LONG_SUCCESS_MSG = "success"; @@ -947,9 +947,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String SUCCESS_PREFIX = "success"; @@ -1080,7 +1080,7 @@ public class AsyncCompletableFutureRetryTopicScenarioTests { static class MyCustomDltProcessor { - final List receivedMsg = new ArrayList<>(); + private final List receivedMsg = Collections.synchronizedList(new ArrayList<>()); MyCustomDltProcessor(KafkaTemplate kafkaTemplate, CountDownLatch latch) { 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 d3c80bf2..fff1e2c6 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 @@ -1,5 +1,5 @@ /* - * Copyright 2024 the original author or authors. + * Copyright 2024-2025 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. @@ -598,9 +598,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - private final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); @KafkaHandler public Mono listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) { @@ -630,9 +630,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - private final List receivedMsgs = new ArrayList<>(); + private final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); @@ -687,9 +687,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + protected final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1); @@ -746,9 +746,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + protected final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String FAIL_PREFIX = "fail"; @@ -811,9 +811,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + protected final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String LONG_SUCCESS_MSG = "success"; @@ -877,9 +877,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + protected final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String LONG_SUCCESS_MSG = "success"; @@ -943,9 +943,9 @@ public class AsyncMonoRetryTopicScenarioTests { @Autowired CountDownLatchContainer container; - protected final List receivedMsgs = new ArrayList<>(); + protected final List receivedMsgs = Collections.synchronizedList(new ArrayList<>()); - private final List receivedTopics = new ArrayList<>(); + private final List receivedTopics = Collections.synchronizedList(new ArrayList<>()); public static final String SUCCESS_PREFIX = "success"; @@ -1076,7 +1076,7 @@ public class AsyncMonoRetryTopicScenarioTests { static class MyCustomDltProcessor { - final List receivedMsg = new ArrayList<>(); + final List receivedMsg = Collections.synchronizedList(new ArrayList<>()); MyCustomDltProcessor(KafkaTemplate kafkaTemplate, CountDownLatch latch) { this.kafkaTemplate = kafkaTemplate;