Fixed message filtering for batch subscription

allegro · Aug 21, 2024 · f8dad25 · f8dad25
1 parent 5d0f48b
commit f8dad25
Showing 1 changed file with 6 additions and 4 deletions.
diff --git a/...s/src/main/java/pl/allegro/tech/hermes/consumers/consumer/batch/MessageBatchReceiver.java b/...s/src/main/java/pl/allegro/tech/hermes/consumers/consumer/batch/MessageBatchReceiver.java
@@ -111,12 +111,14 @@ private Optional<Message> readAndTransform(Subscription subscription, String bat
         if (maybeMessage.isPresent()) {
             Message message = maybeMessage.get();
 
-            Message transformed = messageConverterResolver.converterFor(message, subscription).convert(message, topic);
-            transformed = message().fromMessage(transformed).withData(wrap(subscription, transformed)).build();
+            if (!message.isFiltered()) {
+                Message transformed = messageConverterResolver.converterFor(message, subscription).convert(message, topic);
+                transformed = message().fromMessage(transformed).withData(wrap(subscription, transformed)).build();
 
-            trackers.get(subscription).logInflight(toMessageMetadata(transformed, subscription, batchId));
+                trackers.get(subscription).logInflight(toMessageMetadata(transformed, subscription, batchId));
 
-            return Optional.of(transformed);
+                return Optional.of(transformed);
+            }
         }
         return Optional.empty();
     }