(flink-connector-kafka) branch main updated: [FLINK-39892][metrics] Fix incorrect numBytesOut reported delta after producer reset

[pnowojski]

This is an automated email from the ASF dual-hosted git repository.

pnowojski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git


The following commit(s) were added to refs/heads/main by this push:
     new a88f9a55 [FLINK-39892][metrics] Fix incorrect numBytesOut reported 
delta after producer reset
a88f9a55 is described below

commit a88f9a555cfcd217ed706c057fe3f6af1bd7a29e
Author: Efrat Levitan <elevi...@confluent.io>
AuthorDate: Tue Jun 9 13:38:27 2026 +0300

    [FLINK-39892][metrics] Fix incorrect numBytesOut reported delta after 
producer reset
    
    The following counters
    
    taskmanager.job.task.numBytesOutPerSecond
    taskmanager.job.task.numBytesOut
    taskmanager.job.task.operator.numBytesOutPerSecond
    taskmanager.job.task.operator.numBytesSend
    
    frequently report a value decrement for exactly once kafka sink.
    This is because KafkaWriter#initKafkaMetrics is called per producer and 
resets byteOutMetric but not latestOutgoingByteTotal
    KafkaWriter#registerMetricSync calculates the delta of 
outgoingBytesUntilNow - latestOutgoingByteTotal so every time a new producer is 
spawned, numBytesOutCounter.inc is called with a negative delta, (0 - 
latestOutgoingByteTotal of the former producer)
---
 .../flink/connector/kafka/sink/KafkaWriter.java      |  1 +
 .../connector/kafka/sink/KafkaWriterITCase.java      | 20 +++++++++++++++-----
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git 
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
 
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index 36bd6697..42729fb6 100644
--- 
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ 
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -228,6 +228,7 @@ class KafkaWriter<IN>
         byteOutMetric =
                 MetricUtil.getKafkaMetric(
                         producer.metrics(), KAFKA_PRODUCER_METRICS, 
"outgoing-byte-total");
+        latestOutgoingByteTotal = ((Number) 
byteOutMetric.metricValue()).longValue();
         if (disabledMetrics) {
             return;
         }
diff --git 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index a8ad1ceb..32608ece 100644
--- 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -68,14 +68,15 @@ public class KafkaWriterITCase extends KafkaWriterTestBase {
         assertKafkaMetricNotPresent(guarantee, "register.producer.metrics", 
"false");
     }
 
-    @Test
-    public void testIncreasingRecordBasedCounters() throws Exception {
+    @ParameterizedTest
+    @EnumSource(
+            value = DeliveryGuarantee.class,
+            names = {"AT_LEAST_ONCE", "EXACTLY_ONCE"})
+    public void testIncreasingRecordBasedCounters(DeliveryGuarantee guarantee) 
throws Exception {
         final SinkWriterMetricGroup metricGroup = 
createSinkWriterMetricGroup();
 
         try (final KafkaWriter<Integer> writer =
-                createWriter(
-                        DeliveryGuarantee.AT_LEAST_ONCE,
-                        new SinkInitContext(metricGroup, timeService, null))) {
+                createWriter(guarantee, new SinkInitContext(metricGroup, 
timeService, null))) {
             final Counter numBytesOut = 
metricGroup.getIOMetricGroup().getNumBytesOutCounter();
             final Counter numRecordsOut = 
metricGroup.getIOMetricGroup().getNumRecordsOutCounter();
             final Counter numRecordsOutErrors = 
metricGroup.getNumRecordsOutErrorsCounter();
@@ -107,6 +108,15 @@ public class KafkaWriterITCase extends KafkaWriterTestBase 
{
             assertThat(numRecordsOut.getCount()).isEqualTo(2);
             assertThat(numRecordsOutErrors.getCount()).isEqualTo(0);
             assertThat(numRecordsSendErrors.getCount()).isEqualTo(0);
+
+            // numBytesOut doesn't decrease between producers
+            baselineCount = numBytesOut.getCount();
+            writer.prepareCommit();
+            writer.snapshotState(1);
+            timeService.trigger();
+            assertThat(numBytesOut.getCount())
+                    .as("numBytesOut must increment across producers")
+                    .isGreaterThanOrEqualTo(baselineCount);
         }
     }