Make filtered message receiver always returning a message

allegro · Aug 14, 2024 · d2a4130 · d2a4130
1 parent ffcc75b
commit d2a4130
Show file tree

Hide file tree

Showing 5 changed files with 29 additions and 20 deletions.
diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/Message.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/Message.java
@@ -53,6 +53,8 @@ public class Message implements FilterableMessage {
 
     private long currentMessageBackoff = -1;
 
+    private boolean isFiltered = false;
+
     public Message(String id,
                    String topic,
                    byte[] content,
@@ -122,7 +124,7 @@ public boolean isTtlExceeded(long ttlMillis) {
 
     public synchronized void incrementRetryCounter(Collection<URI> succeededUris) {
         this.retryCounter++;
-        this.succeededUris.addAll(succeededUris.stream().map(URI::toString).collect(toList()));
+        this.succeededUris.addAll(succeededUris.stream().map(URI::toString).toList());
     }
 
     public synchronized int getRetryCounter() {
@@ -206,6 +208,14 @@ public String getSubscription() {
         return subscription;
     }
 
+    public boolean isFiltered() {
+        return isFiltered;
+    }
+
+    public void setFiltered(boolean filtered) {
+        isFiltered = filtered;
+    }
+
     public static class Builder {
         private String id;
         private PartitionOffset partitionOffset;

diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/SerialConsumer.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/SerialConsumer.java
@@ -124,15 +124,19 @@ public void consume(Runnable signalsInterrupt) {
             if (maybeMessage.isPresent()) {
                 Message message = maybeMessage.get();
 
-                if (logger.isDebugEnabled()) {
-                    logger.debug(
-                            "Read message {} partition {} offset {}",
-                            message.getContentType(), message.getPartition(), message.getOffset()
-                    );
+                if (message.isFiltered()) {
+                    profiler.flushMeasurements(ConsumerRun.FILTERED);
+                } else {
+                    if (logger.isDebugEnabled()) {
+                        logger.debug(
+                                "Read message {} partition {} offset {}",
+                                message.getContentType(), message.getPartition(), message.getOffset()
+                        );
+                    }
+
+                    Message convertedMessage = messageConverterResolver.converterFor(message, subscription).convert(message, topic);
+                    sendMessage(convertedMessage, profiler);
                 }
-
-                Message convertedMessage = messageConverterResolver.converterFor(message, subscription).convert(message, topic);
-                sendMessage(convertedMessage, profiler);
             } else {
                 pendingOffsets.releaseSlot();
                 profiler.flushMeasurements(ConsumerRun.EMPTY);
@@ -190,7 +194,7 @@ private void initializeMessageReceiver() {
                 rateLimiter,
                 loadRecorder,
                 metrics,
-                pendingOffsets::markAsFiltered
+                pendingOffsets::markAsProcessed
         );
     }
 

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
@@ -49,10 +49,6 @@ public void markAsProcessed(SubscriptionPartitionOffset subscriptionPartitionOff
         slots.put(subscriptionPartitionOffset, MessageState.PROCESSED);
     }
 
-    public void markAsFiltered(SubscriptionPartitionOffset subscriptionPartitionOffset) {
-        slots.put(subscriptionPartitionOffset, MessageState.FILTERED);
-    }
-
     public boolean tryAcquireSlot(Duration processingInterval) throws InterruptedException {
         if (inflightSemaphore.tryAcquire(processingInterval.toMillis(), TimeUnit.MILLISECONDS)) {
             if (maxPendingOffsetsSemaphore.tryAcquire(processingInterval.toMillis(), TimeUnit.MILLISECONDS)) {

diff --git a/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/profiling/ConsumerRun.java b/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/profiling/ConsumerRun.java
@@ -1,5 +1,5 @@
 package pl.allegro.tech.hermes.consumers.consumer.profiling;
 
 public enum ConsumerRun {
-    EMPTY, DELIVERED, DISCARDED, RETRIED
+    EMPTY, DELIVERED, DISCARDED, RETRIED, FILTERED
 }
diff --git a/...va/pl/allegro/tech/hermes/consumers/consumer/receiver/kafka/FilteringMessageReceiver.java b/...va/pl/allegro/tech/hermes/consumers/consumer/receiver/kafka/FilteringMessageReceiver.java
@@ -35,15 +35,14 @@ public FilteringMessageReceiver(MessageReceiver receiver,
 
     @Override
     public Optional<Message> next() {
-        return receiver.next().map(message ->
-                allow(message) ? message : null
-        );
+        return receiver.next().map(this::filter);
     }
 
-    private boolean allow(Message message) {
+    private Message filter(Message message) {
         FilterResult result = filterChain.apply(message);
         filteredMessageHandler.handle(result, message, subscription);
-        return !result.isFiltered();
+        message.setFiltered(result.isFiltered());
+        return message;
     }
 
     @Override