HADOOP-18567. LogThrottlingHelper: properly trigger dependent recorders in cases of infrequent logging (#5215)

Signed-off-by: Erik Krogen <xkrogen@apache.org>
Co-authored-by: Chengbing Liu <liuchengbing@qiyi.com>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/log/LogThrottlingHelper.java

@@ -262,9 +262,15 @@ public class LogThrottlingHelper {
     if (primaryRecorderName.equals(recorderName) &&
         currentTimeMs - minLogPeriodMs >= lastLogTimestampMs) {
       lastLogTimestampMs = currentTimeMs;
-      for (LoggingAction log : currentLogs.values()) {
+      currentLogs.replaceAll((key, log) -> {
-        log.setShouldLog();
+        LoggingAction newLog = log;
+        if (log.hasLogged()) {
+          // create a fresh log since the old one has already been logged
+          newLog = new LoggingAction(log.getValueCount());
         }
+        newLog.setShouldLog();
+        return newLog;
+      });
     }
     if (currentLog.shouldLog()) {
       currentLog.setHasLogged();
@@ -357,6 +363,10 @@ public class LogThrottlingHelper {
       hasLogged = true;
     }
 
+    private int getValueCount() {
+      return stats.length;
+    }
+
     private void recordValues(double... values) {
       if (values.length != stats.length) {
         throw new IllegalArgumentException("received " + values.length +

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/log/TestLogThrottlingHelper.java

@@ -142,6 +142,18 @@ public class TestLogThrottlingHelper {
     assertTrue(helper.record("bar", 0).shouldLog());
   }
 
+  @Test
+  public void testInfrequentPrimaryAndDependentLoggers() {
+    helper = new LogThrottlingHelper(LOG_PERIOD, "foo", timer);
+
+    assertTrue(helper.record("foo", 0).shouldLog());
+    assertTrue(helper.record("bar", 0).shouldLog());
+
+    // Both should log once the period has elapsed
+    assertTrue(helper.record("foo", LOG_PERIOD).shouldLog());
+    assertTrue(helper.record("bar", LOG_PERIOD).shouldLog());
+  }
+
   @Test
   public void testMultipleLoggersWithValues() {
     helper = new LogThrottlingHelper(LOG_PERIOD, "foo", timer);