Fixed message filtering for batch subscription

allegro · Aug 21, 2024 · e0d04c7 · e0d04c7
1 parent 497351f
commit e0d04c7
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 4 deletions.
diff --git a/...onsumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/MessageState.java b/...onsumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/MessageState.java
@@ -3,5 +3,4 @@
 public enum MessageState {
     INFLIGHT,
     PROCESSED,
-    FILTERED
 }
diff --git a/...umers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/OffsetCommitter.java b/...umers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/OffsetCommitter.java
@@ -95,8 +95,7 @@ public Set<SubscriptionPartitionOffset> calculateOffsetsToBeCommitted(Map<Subscr
         try (HermesTimerContext ignored = timer.time()) {
             List<SubscriptionPartitionOffset> processedOffsets = new ArrayList<>();
             for (Map.Entry<SubscriptionPartitionOffset, MessageState> entry : offsets.entrySet()) {
-                // we consider filtered messages as processed
-                if (entry.getValue() != MessageState.INFLIGHT) {
+                if (entry.getValue() == MessageState.PROCESSED) {
                     processedOffsets.add(entry.getKey());
                 }
             }

diff --git a/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/PendingOffsets.java b/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/PendingOffsets.java
@@ -69,7 +69,7 @@ public Map<SubscriptionPartitionOffset, MessageState> getOffsetsSnapshotAndRelea
 
         for (Map.Entry<SubscriptionPartitionOffset, MessageState> entry : slots.entrySet()) {
             offsetSnapshot.put(entry.getKey(), entry.getValue());
-            if (entry.getValue() != MessageState.INFLIGHT) {
+            if (entry.getValue() == MessageState.PROCESSED) {
                 slots.remove(entry.getKey());
                 permitsReleased++;
             }