NEU

[demos/kafka/chat] / src / main / java / de / juplo / kafka / chat / backend / persistence / kafka / ChatMessageChannel.java

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
index 1925cc8..084986c 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
@@ -227,11 +227,7 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
     return IntStream
         .range(0, numShards)
         .filter(shard -> isShardOwned[shard])
-        .mapToObj(shard -> nextOffset[shard] >= currentOffset[shard])
-        .collect(
-            () -> Boolean.TRUE, // TODO: Boolean is immutable
-            (acc, v) -> Boolean.valueOf(acc && v), // TODO: Boolean is immutable
-            (a, b) -> Boolean.valueOf(a && b)); // TODO: Boolean is immutable
+        .allMatch(shard -> nextOffset[shard] >= currentOffset[shard]);
   }
 
   void pauseAllOwnedPartions()