Fixed bug leading to semaphore leak

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

Hide file tree

Showing 5 changed files with 74 additions and 3 deletions.
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
@@ -190,7 +190,7 @@ private void initializeMessageReceiver() {
                 rateLimiter,
                 loadRecorder,
                 metrics,
-                pendingOffsets::markAsProcessed
+                pendingOffsets::markAsFiltered
         );
     }
 

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
@@ -2,5 +2,6 @@
 
 public enum MessageState {
     INFLIGHT,
-    PROCESSED
+    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,7 +95,8 @@ public Set<SubscriptionPartitionOffset> calculateOffsetsToBeCommitted(Map<Subscr
         try (HermesTimerContext ignored = timer.time()) {
             List<SubscriptionPartitionOffset> processedOffsets = new ArrayList<>();
             for (Map.Entry<SubscriptionPartitionOffset, MessageState> entry : offsets.entrySet()) {
-                if (entry.getValue() == MessageState.PROCESSED) {
+                // we consider filtered messages as processed
+                if (entry.getValue() != MessageState.INFLIGHT) {
                     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
@@ -49,6 +49,10 @@ 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)) {
@@ -72,6 +76,8 @@ public Map<SubscriptionPartitionOffset, MessageState> getOffsetsSnapshotAndRelea
             if (entry.getValue() == MessageState.PROCESSED) {
                 slots.remove(entry.getKey());
                 permitsReleased++;
+            } else if (entry.getValue() == MessageState.FILTERED) {
+                slots.remove(entry.getKey());
             }
         }
         maxPendingOffsetsSemaphore.release(permitsReleased);

diff --git a/...n-tests/src/integrationTest/java/pl/allegro/tech/hermes/integrationtests/MetricsTest.java b/...n-tests/src/integrationTest/java/pl/allegro/tech/hermes/integrationtests/MetricsTest.java
@@ -257,6 +257,69 @@ public void shouldReportMetricForFilteredSubscription() {
         );
     }
 
+    @Test
+    public void shouldNotIncreaseInflightForFilteredSubscription() {
+        // given
+        TestSubscriber subscriber = subscribers.createSubscriber(503);
+        Topic topic = hermes.initHelper().createTopic(topicWithRandomName().build());
+        final Subscription subscription = hermes.initHelper().createSubscription(
+                subscription(topic, "subscription")
+                        .withEndpoint(subscriber.getEndpoint())
+                        .withSubscriptionPolicy(
+                                subscriptionPolicy()
+                                        .withMessageTtl(3600)
+                                        .withInflightSize(1)
+                                        .build()
+                        )
+                        .withFilter(filterMatchingHeaderByPattern("Trace-Id", "^vte.*"))
+                        .build()
+        );
+        TestMessage unfiltered = TestMessage.of("msg", "unfiltered");
+        TestMessage filtered = TestMessage.of("msg", "filtered");
+
+        // when
+        hermes.api().publishUntilSuccess(topic.getQualifiedName(), filtered.body(), header("Trace-Id", "otherTraceId"));
+        hermes.api().publishUntilSuccess(topic.getQualifiedName(), filtered.body(), header("Trace-Id", "otherTraceId"));
+
+        // then
+        waitAtMost(Duration.ofSeconds(10)).untilAsserted(() ->
+                hermes.api().getConsumersMetrics()
+                        .expectStatus()
+                        .isOk()
+                        .expectBody(String.class)
+                        .value((body) -> assertThatMetrics(body)
+                                .contains("hermes_consumers_subscription_inflight")
+                                .withLabels(
+                                        "group", topic.getName().getGroupName(),
+                                        "subscription", subscription.getName(),
+                                        "topic", topic.getName().getName()
+                                )
+                                .withValue(0.0)
+                        )
+        );
+
+        // when
+        hermes.api().publishUntilSuccess(topic.getQualifiedName(), unfiltered.body(), header("Trace-Id", "vte12"));
+        hermes.api().publishUntilSuccess(topic.getQualifiedName(), unfiltered.body(), header("Trace-Id", "vte12"));
+
+        // then
+        waitAtMost(Duration.ofSeconds(10)).untilAsserted(() ->
+                hermes.api().getConsumersMetrics()
+                        .expectStatus()
+                        .isOk()
+                        .expectBody(String.class)
+                        .value((body) -> assertThatMetrics(body)
+                                .contains("hermes_consumers_subscription_inflight")
+                                .withLabels(
+                                        "group", topic.getName().getGroupName(),
+                                        "subscription", subscription.getName(),
+                                        "topic", topic.getName().getName()
+                                )
+                                .withValue(1.0)
+                        )
+        );
+    }
+
     @Test
     public void shouldReportMetricsForSuccessfulBatchDelivery() {
         // given