(flink) branch release-1.20 updated: [FLINK-36260][metrics] Fix numBytesInLocal and numBuffersInLocal being reported as remote

[pnowojski]

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

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


The following commit(s) were added to refs/heads/release-1.20 by this push:
     new 5b78f7fd4bd [FLINK-36260][metrics] Fix numBytesInLocal and 
numBuffersInLocal being reported as remote
5b78f7fd4bd is described below

commit 5b78f7fd4bd092de517a4af6afc83e8ecfe12dbc
Author: Piotr Nowojski <piotr.nowoj...@gmail.com>
AuthorDate: Mon Sep 30 15:30:33 2024 +0200

    [FLINK-36260][metrics] Fix numBytesInLocal and numBuffersInLocal being 
reported as remote
---
 .../partition/consumer/UnknownInputChannel.java    |  4 +-
 .../consumer/UnknownInputChannelTest.java          | 51 ++++++++++++++++++++++
 2 files changed, 53 insertions(+), 2 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
index 6980ed507da..2ff8aa73bcd 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
@@ -196,8 +196,8 @@ class UnknownInputChannel extends InputChannel implements 
ChannelStateHolder {
                 taskEventPublisher,
                 initialBackoff,
                 maxBackoff,
-                metrics.getNumBytesInRemoteCounter(),
-                metrics.getNumBuffersInRemoteCounter(),
+                metrics.getNumBytesInLocalCounter(),
+                metrics.getNumBuffersInLocalCounter(),
                 channelStateWriter == null ? ChannelStateWriter.NO_OP : 
channelStateWriter);
     }
 
diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannelTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannelTest.java
new file mode 100644
index 00000000000..c8b10f4657d
--- /dev/null
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannelTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition.consumer;
+
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link UnknownInputChannel}. */
+class UnknownInputChannelTest {
+    @Test
+    void testMetrics() {
+        SingleInputGateBuilder builder =
+                new SingleInputGateBuilder()
+                        .setNumberOfChannels(1)
+                        .setSingleInputGateIndex(0)
+                        .setResultPartitionType(ResultPartitionType.PIPELINED);
+
+        InputChannelMetrics metrics = new InputChannelMetrics(new 
UnregisteredMetricsGroup());
+        UnknownInputChannel unknownInputChannel =
+                InputChannelBuilder.newBuilder()
+                        .setMetrics(metrics)
+                        .buildUnknownChannel(builder.build());
+        metrics.getNumBuffersInLocalCounter().inc();
+        LocalInputChannel localInputChannel =
+                unknownInputChannel.toLocalInputChannel(new 
ResultPartitionID());
+        assertThat(localInputChannel.numBuffersIn.getCount())
+                .isEqualTo(metrics.getNumBuffersInLocalCounter().getCount());
+    }
+}