Skip to content

Commit 499f52d

Browse files
Wzy19930507Wzy19930507
authored
ConcurrentMessageListenerMockTests cleanup
* `Arrays.asList` to `List.of` in `ConcurrentMessageListenerContainerMockTests` * Other minor cleanup
1 parent 02c6383 commit 499f52d

File tree

1 file changed

+47
-43
lines changed

1 file changed

+47
-43
lines changed

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java

Lines changed: 47 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030

3131
import java.time.Duration;
3232
import java.util.ArrayList;
33-
import java.util.Arrays;
3433
import java.util.Collection;
3534
import java.util.Collections;
3635
import java.util.HashMap;
@@ -132,7 +131,7 @@ else if (event instanceof ConsumerFailedToStartEvent) {
132131
exec.destroy();
133132
}
134133

135-
@SuppressWarnings({ "rawtypes", "unchecked", "deprecation" })
134+
@SuppressWarnings({ "rawtypes", "unchecked" })
136135
@Test
137136
void testCorrectContainerForConsumerError() throws InterruptedException {
138137
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
@@ -202,10 +201,8 @@ void delayedIdleEvent() throws InterruptedException {
202201
containerProperties);
203202
CountDownLatch latch1 = new CountDownLatch(1);
204203
CountDownLatch latch2 = new CountDownLatch(2);
205-
AtomicReference<Long> eventTime = new AtomicReference<>();
206204
container.setApplicationEventPublisher(event -> {
207205
if (event instanceof ListenerContainerIdleEvent) {
208-
eventTime.set(System.currentTimeMillis());
209206
latch1.countDown();
210207
latch2.countDown();
211208
}
@@ -265,7 +262,7 @@ void testSyncRelativeSeeks() throws InterruptedException {
265262
TopicPartition tp1 = new TopicPartition("foo", 1);
266263
TopicPartition tp2 = new TopicPartition("foo", 2);
267264
TopicPartition tp3 = new TopicPartition("foo", 3);
268-
List<TopicPartition> assignments = Arrays.asList(tp0, tp1, tp2, tp3);
265+
List<TopicPartition> assignments = List.of(tp0, tp1, tp2, tp3);
269266
willAnswer(invocation -> {
270267
((ConsumerRebalanceListener) invocation.getArgument(1))
271268
.onPartitionsAssigned(assignments);
@@ -306,7 +303,7 @@ void testAsyncRelativeSeeks() throws InterruptedException {
306303
TopicPartition tp1 = new TopicPartition("foo", 1);
307304
TopicPartition tp2 = new TopicPartition("foo", 2);
308305
TopicPartition tp3 = new TopicPartition("foo", 3);
309-
List<TopicPartition> assignments = Arrays.asList(tp0, tp1, tp2, tp3);
306+
List<TopicPartition> assignments = List.of(tp0, tp1, tp2, tp3);
310307
Map<TopicPartition, List<ConsumerRecord<String, String>>> recordMap = new HashMap<>();
311308
recordMap.put(tp0, Collections.singletonList(new ConsumerRecord("foo", 0, 0, null, "bar")));
312309
recordMap.put(tp1, Collections.singletonList(new ConsumerRecord("foo", 1, 0, null, "bar")));
@@ -365,7 +362,7 @@ void testSyncTimestampSeeks() throws InterruptedException {
365362
TopicPartition tp1 = new TopicPartition("foo", 1);
366363
TopicPartition tp2 = new TopicPartition("foo", 2);
367364
TopicPartition tp3 = new TopicPartition("foo", 3);
368-
List<TopicPartition> assignments = Arrays.asList(tp0, tp1, tp2, tp3);
365+
List<TopicPartition> assignments = List.of(tp0, tp1, tp2, tp3);
369366
willAnswer(invocation -> {
370367
((ConsumerRebalanceListener) invocation.getArgument(1))
371368
.onPartitionsAssigned(assignments);
@@ -412,7 +409,7 @@ void testAsyncTimestampSeeks() throws InterruptedException {
412409
TopicPartition tp1 = new TopicPartition("foo", 1);
413410
TopicPartition tp2 = new TopicPartition("foo", 2);
414411
TopicPartition tp3 = new TopicPartition("foo", 3);
415-
List<TopicPartition> assignments = Arrays.asList(tp0, tp1, tp2, tp3);
412+
List<TopicPartition> assignments = List.of(tp0, tp1, tp2, tp3);
416413
Map<TopicPartition, List<ConsumerRecord<String, String>>> recordMap = new HashMap<>();
417414
recordMap.put(tp0, Collections.singletonList(new ConsumerRecord("foo", 0, 0, null, "bar")));
418415
recordMap.put(tp1, Collections.singletonList(new ConsumerRecord("foo", 1, 0, null, "bar")));
@@ -510,7 +507,9 @@ void testBatchInterceptBeforeTx1() throws InterruptedException {
510507
}
511508

512509
@SuppressWarnings({ "rawtypes", "unchecked" })
513-
void testIntercept(boolean beforeTx, AssignmentCommitOption option, boolean batch) throws InterruptedException {
510+
void testIntercept(boolean beforeTx, @Nullable AssignmentCommitOption option, boolean batch)
511+
throws InterruptedException {
512+
514513
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
515514
final Consumer consumer = mock(Consumer.class);
516515
TopicPartition tp0 = new TopicPartition("foo", 0);
@@ -525,7 +524,7 @@ void testIntercept(boolean beforeTx, AssignmentCommitOption option, boolean batc
525524
Thread.sleep(10);
526525
return firstOrSecondPoll.incrementAndGet() < 3 ? records : empty;
527526
}).given(consumer).poll(any());
528-
List<TopicPartition> assignments = Arrays.asList(tp0);
527+
List<TopicPartition> assignments = List.of(tp0);
529528
willAnswer(invocation -> {
530529
((ConsumerRebalanceListener) invocation.getArgument(1))
531530
.onPartitionsAssigned(assignments);
@@ -678,7 +677,7 @@ void testInterceptInTxNonKafkaTM() throws InterruptedException {
678677
Thread.sleep(10);
679678
return firstOrSecondPoll.incrementAndGet() < 2 ? records : empty;
680679
}).given(consumer).poll(any());
681-
List<TopicPartition> assignments = Arrays.asList(tp0);
680+
List<TopicPartition> assignments = List.of(tp0);
682681
willAnswer(invocation -> {
683682
((ConsumerRebalanceListener) invocation.getArgument(1))
684683
.onPartitionsAssigned(assignments);
@@ -773,7 +772,7 @@ void testNoCommitOnAssignmentWithEarliest() throws InterruptedException {
773772
return records;
774773
}).given(consumer).poll(any());
775774
TopicPartition tp0 = new TopicPartition("foo", 0);
776-
List<TopicPartition> assignments = Arrays.asList(tp0);
775+
List<TopicPartition> assignments = List.of(tp0);
777776
willAnswer(invocation -> {
778777
((ConsumerRebalanceListener) invocation.getArgument(1))
779778
.onPartitionsAssigned(assignments);
@@ -816,7 +815,7 @@ private void testInitialCommitIBasedOnCommitted(boolean committed) throws Interr
816815
return records;
817816
}).given(consumer).poll(any());
818817
TopicPartition tp0 = new TopicPartition("foo", 0);
819-
List<TopicPartition> assignments = Arrays.asList(tp0);
818+
List<TopicPartition> assignments = List.of(tp0);
820819
willAnswer(invocation -> {
821820
((ConsumerRebalanceListener) invocation.getArgument(1))
822821
.onPartitionsAssigned(assignments);
@@ -867,7 +866,7 @@ void removeFromPartitionPauseRequestedWhenNotAssigned() throws InterruptedExcept
867866
return null;
868867
}).given(consumer).pause(any());
869868
TopicPartition tp0 = new TopicPartition("foo", 0);
870-
List<TopicPartition> assignments = Arrays.asList(tp0);
869+
List<TopicPartition> assignments = List.of(tp0);
871870
AtomicReference<ConsumerRebalanceListener> rebal = new AtomicReference<>();
872871
willAnswer(invocation -> {
873872
rebal.set(invocation.getArgument(1));
@@ -913,14 +912,14 @@ void pruneRevokedPartitionsFromRemainingRecordsWhenSeekAfterErrorFalseLegacyAssi
913912
TopicPartition tp1 = new TopicPartition("foo", 1);
914913
TopicPartition tp2 = new TopicPartition("foo", 2);
915914
TopicPartition tp3 = new TopicPartition("foo", 3);
916-
List<TopicPartition> allAssignments = Arrays.asList(tp0, tp1, tp2, tp3);
915+
List<TopicPartition> allAssignments = List.of(tp0, tp1, tp2, tp3);
917916
Map<TopicPartition, List<ConsumerRecord<String, String>>> allRecordMap = new HashMap<>();
918917
allRecordMap.put(tp0, Collections.singletonList(new ConsumerRecord("foo", 0, 0, null, "bar")));
919918
allRecordMap.put(tp1, Collections.singletonList(new ConsumerRecord("foo", 1, 0, null, "bar")));
920919
allRecordMap.put(tp2, Collections.singletonList(new ConsumerRecord("foo", 2, 0, null, "bar")));
921920
allRecordMap.put(tp3, Collections.singletonList(new ConsumerRecord("foo", 3, 0, null, "bar")));
922921
ConsumerRecords allRecords = new ConsumerRecords<>(allRecordMap);
923-
List<TopicPartition> afterRevokeAssignments = Arrays.asList(tp1, tp3);
922+
List<TopicPartition> afterRevokeAssignments = List.of(tp1, tp3);
924923
Map<TopicPartition, List<ConsumerRecord<String, String>>> afterRevokeRecordMap = new HashMap<>();
925924
afterRevokeRecordMap.put(tp1, Collections.singletonList(new ConsumerRecord("foo", 1, 0, null, "bar")));
926925
afterRevokeRecordMap.put(tp3, Collections.singletonList(new ConsumerRecord("foo", 3, 0, null, "bar")));
@@ -981,10 +980,11 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
981980
Thread.sleep(50);
982981
pollLatch.countDown();
983982
switch (pollPhase.getAndIncrement()) {
984-
case 0:
983+
case 0 -> {
985984
rebal.get().onPartitionsAssigned(allAssignments);
986985
return allRecords;
987-
case 1:
986+
}
987+
case 1 -> {
988988
rebal.get().onPartitionsRevoked(allAssignments);
989989
rebal.get().onPartitionsAssigned(afterRevokeAssignments);
990990
rebalLatch.countDown();
@@ -993,11 +993,13 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
993993
return ConsumerRecords.empty();
994994
}
995995
return afterRevokeRecords;
996-
default:
996+
}
997+
default -> {
997998
if (paused.get()) {
998999
return ConsumerRecords.empty();
9991000
}
10001001
return afterRevokeRecords;
1002+
}
10011003
}
10021004
}).given(consumer).poll(any());
10031005
container.start();
@@ -1025,7 +1027,7 @@ void pruneRevokedPartitionsFromRemainingRecordsWhenSeekAfterErrorFalseCoopAssign
10251027
TopicPartition tp1 = new TopicPartition("foo", 1);
10261028
TopicPartition tp2 = new TopicPartition("foo", 2);
10271029
TopicPartition tp3 = new TopicPartition("foo", 3);
1028-
List<TopicPartition> allAssignments = Arrays.asList(tp0, tp1, tp2, tp3);
1030+
List<TopicPartition> allAssignments = List.of(tp0, tp1, tp2, tp3);
10291031
Map<TopicPartition, List<ConsumerRecord<String, String>>> allRecordMap = new LinkedHashMap<>();
10301032
ConsumerRecord record0 = new ConsumerRecord("foo", 0, 0, null, "bar");
10311033
ConsumerRecord record1 = new ConsumerRecord("foo", 1, 0, null, "bar");
@@ -1034,7 +1036,7 @@ void pruneRevokedPartitionsFromRemainingRecordsWhenSeekAfterErrorFalseCoopAssign
10341036
allRecordMap.put(tp2, Collections.singletonList(new ConsumerRecord("foo", 2, 0, null, "bar")));
10351037
allRecordMap.put(tp3, Collections.singletonList(new ConsumerRecord("foo", 3, 0, null, "bar")));
10361038
ConsumerRecords allRecords = new ConsumerRecords<>(allRecordMap);
1037-
List<TopicPartition> revokedAssignments = Arrays.asList(tp0, tp2);
1039+
List<TopicPartition> revokedAssignments = List.of(tp0, tp2);
10381040
AtomicInteger pollPhase = new AtomicInteger();
10391041

10401042
Consumer consumer = mock(Consumer.class);
@@ -1046,9 +1048,7 @@ void pruneRevokedPartitionsFromRemainingRecordsWhenSeekAfterErrorFalseCoopAssign
10461048
return null;
10471049
}).given(consumer).subscribe(any(Collection.class), any());
10481050
CountDownLatch pauseLatch = new CountDownLatch(1);
1049-
AtomicBoolean paused = new AtomicBoolean();
10501051
willAnswer(inv -> {
1051-
paused.set(true);
10521052
pauseLatch.countDown();
10531053
return null;
10541054
}).given(consumer).pause(any());
@@ -1089,17 +1089,20 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
10891089
Thread.sleep(50);
10901090
pollLatch.countDown();
10911091
switch (pollPhase.getAndIncrement()) {
1092-
case 0:
1092+
case 0 -> {
10931093
rebal.get().onPartitionsAssigned(allAssignments);
10941094
return allRecords;
1095-
case 1:
1095+
}
1096+
case 1 -> {
10961097
rebal.get().onPartitionsRevoked(revokedAssignments);
10971098
rebal.get().onPartitionsAssigned(Collections.emptyList());
10981099
rebalLatch.countDown();
10991100
continueLatch.await(10, TimeUnit.SECONDS);
11001101
return ConsumerRecords.empty();
1101-
default:
1102+
}
1103+
default -> {
11021104
return ConsumerRecords.empty();
1105+
}
11031106
}
11041107
}).given(consumer).poll(any());
11051108
container.start();
@@ -1130,14 +1133,14 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
11301133
void pruneRevokedPartitionsFromPendingOutOfOrderCommitsLegacyAssignor() throws InterruptedException {
11311134
TopicPartition tp0 = new TopicPartition("foo", 0);
11321135
TopicPartition tp1 = new TopicPartition("foo", 1);
1133-
List<TopicPartition> allAssignments = Arrays.asList(tp0, tp1);
1136+
List<TopicPartition> allAssignments = List.of(tp0, tp1);
11341137
Map<TopicPartition, List<ConsumerRecord<String, String>>> allRecordMap = new HashMap<>();
11351138
allRecordMap.put(tp0,
11361139
List.of(new ConsumerRecord("foo", 0, 0, null, "bar"), new ConsumerRecord("foo", 0, 1, null, "bar")));
11371140
allRecordMap.put(tp1,
11381141
List.of(new ConsumerRecord("foo", 1, 0, null, "bar"), new ConsumerRecord("foo", 1, 1, null, "bar")));
11391142
ConsumerRecords allRecords = new ConsumerRecords<>(allRecordMap);
1140-
List<TopicPartition> afterRevokeAssignments = Arrays.asList(tp1);
1143+
List<TopicPartition> afterRevokeAssignments = List.of(tp1);
11411144
AtomicInteger pollPhase = new AtomicInteger();
11421145

11431146
Consumer consumer = mock(Consumer.class);
@@ -1149,9 +1152,7 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsLegacyAssignor() throws I
11491152
return null;
11501153
}).given(consumer).subscribe(any(Collection.class), any());
11511154
CountDownLatch pauseLatch = new CountDownLatch(1);
1152-
AtomicBoolean paused = new AtomicBoolean();
11531155
willAnswer(inv -> {
1154-
paused.set(true);
11551156
pauseLatch.countDown();
11561157
return null;
11571158
}).given(consumer).pause(any());
@@ -1173,17 +1174,20 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsLegacyAssignor() throws I
11731174
Thread.sleep(50);
11741175
pollLatch.countDown();
11751176
switch (pollPhase.getAndIncrement()) {
1176-
case 0:
1177+
case 0 -> {
11771178
rebal.get().onPartitionsAssigned(allAssignments);
11781179
return allRecords;
1179-
case 1:
1180+
}
1181+
case 1 -> {
11801182
rebal.get().onPartitionsRevoked(allAssignments);
11811183
rebal.get().onPartitionsAssigned(afterRevokeAssignments);
11821184
rebalLatch.countDown();
11831185
continueLatch.await(10, TimeUnit.SECONDS);
11841186
return ConsumerRecords.empty();
1185-
default:
1187+
}
1188+
default -> {
11861189
return ConsumerRecords.empty();
1190+
}
11871191
}
11881192
}).given(consumer).poll(any());
11891193
container.start();
@@ -1208,14 +1212,13 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsLegacyAssignor() throws I
12081212
void pruneRevokedPartitionsFromPendingOutOfOrderCommitsCoopAssignor() throws InterruptedException {
12091213
TopicPartition tp0 = new TopicPartition("foo", 0);
12101214
TopicPartition tp1 = new TopicPartition("foo", 1);
1211-
List<TopicPartition> allAssignments = Arrays.asList(tp0, tp1);
1215+
List<TopicPartition> allAssignments = List.of(tp0, tp1);
12121216
Map<TopicPartition, List<ConsumerRecord<String, String>>> allRecordMap = new HashMap<>();
12131217
allRecordMap.put(tp0,
12141218
List.of(new ConsumerRecord("foo", 0, 0, null, "bar"), new ConsumerRecord("foo", 0, 1, null, "bar")));
12151219
allRecordMap.put(tp1,
12161220
List.of(new ConsumerRecord("foo", 1, 0, null, "bar"), new ConsumerRecord("foo", 1, 1, null, "bar")));
12171221
ConsumerRecords allRecords = new ConsumerRecords<>(allRecordMap);
1218-
List<TopicPartition> afterRevokeAssignments = Arrays.asList(tp1);
12191222
AtomicInteger pollPhase = new AtomicInteger();
12201223

12211224
Consumer consumer = mock(Consumer.class);
@@ -1227,9 +1230,7 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsCoopAssignor() throws Int
12271230
return null;
12281231
}).given(consumer).subscribe(any(Collection.class), any());
12291232
CountDownLatch pauseLatch = new CountDownLatch(1);
1230-
AtomicBoolean paused = new AtomicBoolean();
12311233
willAnswer(inv -> {
1232-
paused.set(true);
12331234
pauseLatch.countDown();
12341235
return null;
12351236
}).given(consumer).pause(any());
@@ -1251,17 +1252,20 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsCoopAssignor() throws Int
12511252
Thread.sleep(50);
12521253
pollLatch.countDown();
12531254
switch (pollPhase.getAndIncrement()) {
1254-
case 0:
1255+
case 0 -> {
12551256
rebal.get().onPartitionsAssigned(allAssignments);
12561257
return allRecords;
1257-
case 1:
1258+
}
1259+
case 1 -> {
12581260
rebal.get().onPartitionsRevoked(List.of(tp0));
12591261
rebal.get().onPartitionsAssigned(List.of(new TopicPartition("foo", 2)));
12601262
rebalLatch.countDown();
12611263
continueLatch.await(10, TimeUnit.SECONDS);
12621264
return ConsumerRecords.empty();
1263-
default:
1265+
}
1266+
default -> {
12641267
return ConsumerRecords.empty();
1268+
}
12651269
}
12661270
}).given(consumer).poll(any());
12671271
container.start();
@@ -1287,7 +1291,7 @@ private AcknowledgingMessageListener ackOffset1() {
12871291

12881292
@Override
12891293
public void onMessage(ConsumerRecord rec, @Nullable Acknowledgment ack) {
1290-
if (rec.offset() == 1) {
1294+
if (rec.offset() == 1 && ack != null) {
12911295
ack.acknowledge();
12921296
}
12931297
}
@@ -1301,7 +1305,7 @@ public void onMessage(Object data) {
13011305

13021306
public static class TestMessageListener1 implements MessageListener<String, String>, ConsumerSeekAware {
13031307

1304-
private static ThreadLocal<ConsumerSeekCallback> callbacks = new ThreadLocal<>();
1308+
private static final ThreadLocal<ConsumerSeekCallback> callbacks = new ThreadLocal<>();
13051309

13061310
CountDownLatch latch = new CountDownLatch(1);
13071311

@@ -1337,7 +1341,7 @@ public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekC
13371341

13381342
public static class TestMessageListener2 implements MessageListener<String, String>, ConsumerSeekAware {
13391343

1340-
private static ThreadLocal<ConsumerSeekCallback> callbacks = new ThreadLocal<>();
1344+
private static final ThreadLocal<ConsumerSeekCallback> callbacks = new ThreadLocal<>();
13411345

13421346
CountDownLatch latch = new CountDownLatch(1);
13431347

0 commit comments

Comments
 (0)