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

Fix async tests for Collections.synchronizedList()

Browse Source 
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
This commit is contained in:
Artem Bilan
2025-03-06 15:55:10 -05:00
parent ccf466606c
commit c58e1c3b37
2 changed files with 32 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java
View File

@@ -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<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
@KafkaHandler
public CompletableFuture<Void> listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
@@ -634,9 +634,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests {
@Autowired
CountDownLatchContainer container;
private final List<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1);
@@ -691,9 +691,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests {
@Autowired
CountDownLatchContainer container;
protected final List<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1);
@@ -750,9 +750,9 @@ public class AsyncCompletableFutureRetryTopicScenarioTests {
@Autowired
CountDownLatchContainer container;
protected final List<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> 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<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> 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<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> 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<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
public static final String SUCCESS_PREFIX = "success";
@@ -1080,7 +1080,7 @@ public class AsyncCompletableFutureRetryTopicScenarioTests {
static class MyCustomDltProcessor {
final List<String> receivedMsg = new ArrayList<>();
private final List<String> receivedMsg = Collections.synchronizedList(new ArrayList<>());
MyCustomDltProcessor(KafkaTemplate<String, String> kafkaTemplate,
CountDownLatch latch) {

32
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java
View File

@@ -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<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
@KafkaHandler
public Mono<Void> listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
@@ -630,9 +630,9 @@ public class AsyncMonoRetryTopicScenarioTests {
@Autowired
CountDownLatchContainer container;
private final List<String> receivedMsgs = new ArrayList<>();
private final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1);
@@ -687,9 +687,9 @@ public class AsyncMonoRetryTopicScenarioTests {
@Autowired
CountDownLatchContainer container;
protected final List<String> receivedMsgs = new ArrayList<>();
protected final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
private CountDownLatch firstRetryFailMsgLatch = new CountDownLatch(1);
@@ -746,9 +746,9 @@ public class AsyncMonoRetryTopicScenarioTests {
@Autowired
CountDownLatchContainer container;
protected final List<String> receivedMsgs = new ArrayList<>();
protected final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> 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<String> receivedMsgs = new ArrayList<>();
protected final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> 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<String> receivedMsgs = new ArrayList<>();
protected final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> 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<String> receivedMsgs = new ArrayList<>();
protected final List<String> receivedMsgs = Collections.synchronizedList(new ArrayList<>());
private final List<String> receivedTopics = new ArrayList<>();
private final List<String> receivedTopics = Collections.synchronizedList(new ArrayList<>());
public static final String SUCCESS_PREFIX = "success";
@@ -1076,7 +1076,7 @@ public class AsyncMonoRetryTopicScenarioTests {
static class MyCustomDltProcessor {
final List<String> receivedMsg = new ArrayList<>();
final List<String> receivedMsg = Collections.synchronizedList(new ArrayList<>());
MyCustomDltProcessor(KafkaTemplate<String, String> kafkaTemplate, CountDownLatch latch) {
this.kafkaTemplate = kafkaTemplate;