Added OffsetsSlotsTest

allegro · Jul 15, 2024 · 6538eeb · 6538eeb
1 parent dccdef8
commit 6538eeb
Show file tree

Hide file tree

Showing 4 changed files with 1 addition and 203 deletions.
diff --git a/...onsumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/OffsetsSlots.java b/...onsumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/offset/OffsetsSlots.java
@@ -20,7 +20,7 @@ public class OffsetsSlots {
     public OffsetsSlots(SubscriptionName subscriptionName, MetricsFacade metrics, int inflightQueueSize, int offsetQueueSize) {
         this.totalOffsetsCountSemaphore = new Semaphore(offsetQueueSize);
         this.inflightSemaphore = new AdjustableSemaphore(inflightQueueSize);
-        metrics.subscriptions().registerOffsetsQueueGauge(subscriptionName, totalOffsetsCountSemaphore,  slots -> (double) slots.availablePermits() / offsetQueueSize);
+        metrics.subscriptions().registerOffsetsQueueGauge(subscriptionName, totalOffsetsCountSemaphore, slots -> (offsetQueueSize - (double) slots.availablePermits()) / offsetQueueSize);
     }
 
     public void setInflightSize(int inflightQueueSize) {

diff --git a/.../src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/ConsumerPrototype.java b/.../src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/ConsumerPrototype.java
diff --git a/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/OffsetState.java b/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/OffsetState.java
diff --git a/...mers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/PendingOffset.java b/...mers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/PendingOffset.java