From 97e4d5030b60dc195eb2b35634a83eaaf383616e Mon Sep 17 00:00:00 2001
From: huhaiyang <huhaiyang@didichuxing.com>
Date: Mon, 8 Nov 2021 20:45:43 +0800
Subject: [PATCH 1/3] HADOOP-17995. Stale record should be remove when
 DataNodePeerMetrics#dumpSendPacketDownstreamAvgInfoAsJson

---
 .../hadoop/metrics2/lib/MutableRollingAverages.java    |  7 +++++--
 .../hdfs/server/datanode/TestDataNodePeerMetrics.java  | 10 ++++++++++
 2 files changed, 15 insertions(+), 2 deletions(-)

diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
index 193ed0f71d7c7..3217add651476 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
@@ -179,8 +179,11 @@ public void snapshot(MetricsRecordBuilder builder, boolean all) {
         long totalCount = 0;
 
         for (final SumAndCount sumAndCount : entry.getValue()) {
-          totalCount += sumAndCount.getCount();
-          totalSum += sumAndCount.getSum();
+          if (Time.monotonicNow() - sumAndCount.getSnapshotTimeStamp()
+              < recordValidityMs) {
+            totalCount += sumAndCount.getCount();
+            totalSum += sumAndCount.getSum();
+          }
         }
 
         if (totalCount != 0) {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodePeerMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodePeerMetrics.java
index 41fb41f48005a..9ea8a08a7629c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodePeerMetrics.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodePeerMetrics.java
@@ -122,10 +122,16 @@ public void testRemoveStaleRecord() throws Exception {
     GenericTestUtils.waitFor(
         () -> rollingAverages.getStats(numSamples).size() > 0, 500, 5000);
     assertEquals(3, rollingAverages.getStats(numSamples).size());
+    String json = peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson();
+    for (String peerAddr : peerAddrList) {
+      assertThat(json, containsString(peerAddr));
+    }
     /* wait for stale report to be removed */
     GenericTestUtils.waitFor(
         () -> rollingAverages.getStats(numSamples).isEmpty(), 500, 10000);
     assertEquals(0, rollingAverages.getStats(numSamples).size());
+    json = peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson();
+    assertEquals("{}", json);
 
     /* dn can report peer metrics normally when it added back to cluster */
     for (String peerAddr : peerAddrList) {
@@ -138,6 +144,10 @@ public void testRemoveStaleRecord() throws Exception {
     GenericTestUtils.waitFor(
         () -> rollingAverages.getStats(numSamples).size() > 0, 500, 10000);
     assertEquals(3, rollingAverages.getStats(numSamples).size());
+    json = peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson();
+    for (String peerAddr : peerAddrList) {
+      assertThat(json, containsString(peerAddr));
+    }
   }
 
   /**

From aff0c48846ff81d90a19e6da4aee879c66710f5a Mon Sep 17 00:00:00 2001
From: huhaiyang <huhaiyang@didichuxing.com>
Date: Tue, 9 Nov 2021 11:34:09 +0800
Subject: [PATCH 2/3] HADOOP-17995. Fix

---
 .../org/apache/hadoop/metrics2/lib/MutableRollingAverages.java  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
index 3217add651476..aa4d4b9ca0c64 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
@@ -167,7 +167,7 @@ synchronized void replaceScheduledTask(int windows, long interval,
   }
 
   @Override
-  public void snapshot(MetricsRecordBuilder builder, boolean all) {
+  public synchronized void snapshot(MetricsRecordBuilder builder, boolean all) {
     if (all || changed()) {
       for (final Entry<String, LinkedBlockingDeque<SumAndCount>> entry
           : averages.entrySet()) {

From a9fcd6aa932585c4ca260836bd36c67a58fbbfd7 Mon Sep 17 00:00:00 2001
From: huhaiyang <huhaiyang@didichuxing.com>
Date: Fri, 12 Nov 2021 15:05:37 +0800
Subject: [PATCH 3/3] HADOOP-17995. Fix

---
 .../org/apache/hadoop/metrics2/lib/MutableRollingAverages.java  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
index aa4d4b9ca0c64..3217add651476 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
@@ -167,7 +167,7 @@ synchronized void replaceScheduledTask(int windows, long interval,
   }
 
   @Override
-  public synchronized void snapshot(MetricsRecordBuilder builder, boolean all) {
+  public void snapshot(MetricsRecordBuilder builder, boolean all) {
     if (all || changed()) {
       for (final Entry<String, LinkedBlockingDeque<SumAndCount>> entry
           : averages.entrySet()) {