Use messageBatch.getUnfilteredMessageCount() to calculate the fetched…

… batch size
apache · Mar 4, 2024 · d096cd6 · d096cd6
1 parent 88d7234
commit d096cd6
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/...src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -429,7 +429,7 @@ protected boolean consumeLoop()
             _partitionGroupConsumer.fetchMessages(_currentOffset, null, _streamConfig.getFetchTimeoutMillis());
         //track realtime rows fetched on a table level. This included valid + invalid rows
         _serverMetrics.addMeteredTableValue(_clientId, ServerMeter.REALTIME_ROWS_FETCHED,
-            messageBatch.getMessageCount());
+            messageBatch.getUnfilteredMessageCount());
         if (_segmentLogger.isDebugEnabled()) {
           _segmentLogger.debug("message batch received. filtered={} unfiltered={} endOfPartitionGroup={}",
               messageBatch.getMessageCount(), messageBatch.getUnfilteredMessageCount(),