[Fix] Avoid to increment batch counts for empty batches

Author: andsel

logstash-core/src/main/java/org/logstash/execution/QueueReadClientBatchMetrics.java

@@ -38,6 +38,11 @@ public void setupMetrics(AbstractNamespacedMetricExt namespacedMetric) {
     }
 
     public void updateBatchMetrics(QueueBatch batch) {
+        if (batch.events().isEmpty()) {
+            // avoid to increment batch count for empty batches
+            return;
+        }
+
         if (batchMetricMode != QueueFactoryExt.BatchMetricMode.DISABLED) {
             boolean updateMetric = true;
             if (batchMetricMode == QueueFactoryExt.BatchMetricMode.MINIMAL) {

logstash-core/src/test/java/org/logstash/ext/JrubyMemoryReadClientExtTest.java

@@ -104,6 +104,17 @@ public void givenNonEmptyQueueWhenBatchIsReadAndMetricIsDisabledThenBatchCounter
         assertEquals(0L, batchCounter.getValue().longValue());
     }
 
+    @Test
+    public void givenEmptyQueueWhenEmptyBatchIsReadAndMetricIsFullyCollectedThenBatchCounterMetricIsNotUpdated() throws InterruptedException {
+        final JrubyMemoryReadClientExt client = JrubyMemoryReadClientExt.create(queue, 5, 50,
+                QueueFactoryExt.BatchMetricMode.FULL);
+        client.setPipelineMetric(metric);
+
+        final QueueBatch batch = client.readBatch();
+        assertEquals(0, batch.filteredSize());
+        assertEquals(0L, batchCounter.getValue().longValue());
+    }
+
     @Test
     public void givenNonEmptyQueueWhenBatchIsReadThenBatchByteSizeMetricIsUpdated() throws InterruptedException {
         final long expectedBatchByteSize = testEvent.getEvent().estimateMemory();