List<String> actual = new CopyOnWriteArrayList<>();
ExecutorService service = Executors.newSingleThreadExecutor();
Future<?> consumingTask = service.submit(() -> {
while (!Thread.currentThread().isInterrupted()) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> rec : records) {
actual.add(rec.value());
}
}
});
...
try {
Awaitility.await().atMost(5, Seconds)
.until(() -> List.of("A", "B").equals(actual));
} finally {
consumingTask.cancel(true);
service.awaitTermintion(200, MILLISECONDS);
}
Comments
0 B
|👍
/👎