HDFS-17953. DataNode IBR pendingIBRs grows unbounded when a NameNode is unreachable, causing OOM

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -1112,6 +1112,10 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
       = "dfs.blockreport.incremental.intervalMsec";
   public static final long    DFS_BLOCKREPORT_INCREMENTAL_INTERVAL_MSEC_DEFAULT
       = 0;
+  public static final String  DFS_DATANODE_IBR_MAX_PENDING_BLOCKS_KEY
+      = "dfs.datanode.ibr.max.pending.blocks";
+  public static final long    DFS_DATANODE_IBR_MAX_PENDING_BLOCKS_DEFAULT
+      = 1000000;
   public static final String  DFS_BLOCKREPORT_INTERVAL_MSEC_KEY = "dfs.blockreport.intervalMsec";
   public static final long    DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT = 6 * 60 * 60 * 1000;
   public static final String  DFS_BLOCKREPORT_INITIAL_DELAY_KEY = "dfs.blockreport.initialDelay";

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java

@@ -141,6 +141,7 @@ enum RunningState {
     this.dnConf = dn.getDnConf();
     this.ibrManager = new IncrementalBlockReportManager(
         dnConf.ibrInterval,
+        dnConf.ibrMaxPendingBlocks,
         dn.getMetrics());
     prevBlockReportId = ThreadLocalRandom.current().nextLong();
     fullBlockReportLeaseId = 0;
@@ -762,6 +763,14 @@ private void offerService() throws Exception {
               bpos.getBlockPoolId(), getRpcMetricSuffix());
         }
 
+        // Guard against unbounded IBR growth when this NameNode is
+        // unreachable: if the queue was cleared to prevent OOM, schedule a
+        // full block report so the NameNode gets a consistent view once it
+        // becomes reachable again.
+        if (ibrManager.clearIBRsIfNeeded()) {
+          scheduler.forceFullBlockReportNow();
+        }
+
         List<DatanodeCommand> cmds = null;
         boolean forceFullBr =
             scheduler.forceFullBlockReport.getAndSet(false);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java

@@ -113,6 +113,7 @@ public class DNConf {
   volatile boolean diskStatsEnabled;
   volatile long outliersReportIntervalMs;
   final long ibrInterval;
+  final long ibrMaxPendingBlocks;
   volatile long initialBlockReportDelayMs;
   volatile long cacheReportInterval;
   private volatile long datanodeSlowIoWarningThresholdMs;
@@ -206,6 +207,9 @@ public DNConf(final Configurable dn) {
     this.ibrInterval = getConf().getLong(
         DFSConfigKeys.DFS_BLOCKREPORT_INCREMENTAL_INTERVAL_MSEC_KEY,
         DFSConfigKeys.DFS_BLOCKREPORT_INCREMENTAL_INTERVAL_MSEC_DEFAULT);
+    this.ibrMaxPendingBlocks = getConf().getLong(
+        DFSConfigKeys.DFS_DATANODE_IBR_MAX_PENDING_BLOCKS_KEY,
+        DFSConfigKeys.DFS_DATANODE_IBR_MAX_PENDING_BLOCKS_DEFAULT);
     this.blockReportSplitThreshold = getConf().getLong(
         DFS_BLOCKREPORT_SPLIT_THRESHOLD_KEY,
         DFS_BLOCKREPORT_SPLIT_THRESHOLD_DEFAULT);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/IncrementalBlockReportManager.java

@@ -127,6 +127,13 @@ int putMissing(ReceivedDeletedBlockInfo[] rdbis) {
   private final Map<DatanodeStorage, PerStorageIBR> pendingIBRs
       = Maps.newHashMap();
 
+  /**
+   * Total number of pending block entries across all storages. Maintained
+   * incrementally so that the size can be checked in O(1) on the hot path
+   * ({@link #addRDBI}) without iterating over all storages.
+   */
+  private long pendingBlockCount = 0;
+
   /**
    * If this flag is set then an IBR will be sent by the actor
    * thread after waiting for the IBR timer to elapse.
@@ -138,12 +145,22 @@ int putMissing(ReceivedDeletedBlockInfo[] rdbis) {
 
   /** The timestamp of the last IBR. */
   private volatile long lastIBR;
+
+  /**
+   * Maximum number of pending block entries allowed before the queue is
+   * cleared to prevent the DataNode from running out of memory when the
+   * target NameNode is unreachable. 0 disables the check.
+   */
+  private final long maxPendingBlocks;
+
   private DataNodeMetrics dnMetrics;
 
   IncrementalBlockReportManager(
       final long ibrInterval,
+      final long maxPendingBlocks,
       final DataNodeMetrics dnMetrics) {
     this.ibrInterval = ibrInterval;
+    this.maxPendingBlocks = maxPendingBlocks;
     this.lastIBR = monotonicNow() - ibrInterval;
     this.dnMetrics = dnMetrics;
   }
@@ -175,6 +192,8 @@ private synchronized StorageReceivedDeletedBlocks[] generateIBRs() {
         reports.add(new StorageReceivedDeletedBlocks(entry.getKey(), rdbi));
       }
     }
+    // All entries have been drained from the map.
+    pendingBlockCount = 0;
 
     /* set blocks to zero */
     this.dnMetrics.resetBlocksInPendingIBR();
@@ -185,7 +204,10 @@ private synchronized StorageReceivedDeletedBlocks[] generateIBRs() {
 
   private synchronized void putMissing(StorageReceivedDeletedBlocks[] reports) {
     for (StorageReceivedDeletedBlocks r : reports) {
-      pendingIBRs.get(r.getStorage()).putMissing(r.getBlocks());
+      final PerStorageIBR perStorage = pendingIBRs.get(r.getStorage());
+      if (perStorage != null) {
+        pendingBlockCount += perStorage.putMissing(r.getBlocks());
+      }
     }
     if (reports.length > 0) {
       readyToSend = true;
@@ -253,10 +275,12 @@ synchronized void addRDBI(ReceivedDeletedBlockInfo rdbi,
     // There may only be one such entry.
     for (PerStorageIBR perStorage : pendingIBRs.values()) {
       if (perStorage.remove(rdbi.getBlock()) != null) {
+        pendingBlockCount--;
         break;
       }
     }
     getPerStorageIBR(storage).put(rdbi);
+    pendingBlockCount++;
   }
 
   synchronized void notifyNamenodeBlock(ReceivedDeletedBlockInfo rdbi,
@@ -296,12 +320,46 @@ synchronized void triggerDeletionReportForTests() {
     }
   }
 
-  void clearIBRs() {
+  synchronized void clearIBRs() {
     pendingIBRs.clear();
+    pendingBlockCount = 0;
+  }
+
+  /**
+   * Clear the pending IBR queue if it has grown beyond
+   * {@code maxPendingBlocks}. This bounds the DataNode's heap footprint when
+   * the target NameNode is unreachable and the queue would otherwise grow
+   * without bound (for example, a NameNode id listed in
+   * {@code dfs.ha.namenodes.<nsId>} whose RPC address is not configured).
+   *
+   * The caller is responsible for triggering a Full Block Report after a clear
+   * so that the NameNode gets a complete, consistent view once it becomes
+   * reachable again.
+   *
+   * @return true if the queue was cleared; false otherwise.
+   */
+  synchronized boolean clearIBRsIfNeeded() {
+    if (maxPendingBlocks <= 0 || pendingBlockCount < maxPendingBlocks) {
+      return false;
+    }
+    LOG.warn("Clearing {} pending IBR block entries because the count reached "
+        + "the limit {}. The target NameNode appears to be unreachable; a "
+        + "Full Block Report will be sent to resync once it is reachable "
+        + "again.", pendingBlockCount, maxPendingBlocks);
+    clearIBRs();
+    return true;
+  }
+
+  /**
+   * @return the number of pending block entries across all storages.
+   */
+  @VisibleForTesting
+  synchronized long getPendingBlockCount() {
+    return pendingBlockCount;
   }
 
   @VisibleForTesting
-  int getPendingIBRSize() {
+  synchronized int getPendingIBRSize() {
     return pendingIBRs.size();
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -4508,6 +4508,21 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.datanode.ibr.max.pending.blocks</name>
+  <value>1000000</value>
+  <description>
+    Maximum number of pending block entries allowed in the IBR
+    (Incremental Block Report) queue per NameNode. When this limit is
+    reached, the queue is cleared to bound the DataNode's heap footprint,
+    and a Full Block Report is scheduled to resync with the NameNode. This
+    protects against unbounded memory growth (and eventual OOM) when a
+    NameNode is unreachable, for example a NameNode id listed in
+    dfs.ha.namenodes.&lt;nsId&gt; whose RPC address is not configured.
+    Set to 0 to disable.
+  </description>
+</property>
+
 <property>
   <name>dfs.checksum.type</name>
   <value>CRC32C</value>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReportManager.java

@@ -0,0 +1,133 @@
+/**
+ * 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.hadoop.hdfs.server.datanode;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.server.datanode.metrics.DataNodeMetrics;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
+import org.apache.hadoop.hdfs.server.protocol.ReceivedDeletedBlockInfo;
+import org.apache.hadoop.hdfs.server.protocol.ReceivedDeletedBlockInfo.BlockStatus;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link IncrementalBlockReportManager}'s OOM protection: the
+ * pending-block-count cap enforced via
+ * {@link IncrementalBlockReportManager#clearIBRsIfNeeded()}.
+ */
+public class TestIncrementalBlockReportManager {
+
+  private DataNodeMetrics mockMetrics;
+  private DatanodeStorage storage;
+
+  @BeforeEach
+  public void setUp() {
+    mockMetrics = mock(DataNodeMetrics.class);
+    storage = new DatanodeStorage("storage-1");
+  }
+
+  private void addBlocks(IncrementalBlockReportManager ibr,
+      DatanodeStorage st, int startId, int count) {
+    for (int i = 0; i < count; i++) {
+      Block block = new Block(startId + i, 1024, 1000 + startId + i);
+      ibr.addRDBI(new ReceivedDeletedBlockInfo(
+          block, BlockStatus.RECEIVED_BLOCK, null), st);
+    }
+  }
+
+  /**
+   * The pending block counter must be maintained accurately (in O(1)) as
+   * blocks are added, deduplicated and cleared.
+   */
+  @Test
+  public void testPendingBlockCountAccounting() {
+    IncrementalBlockReportManager ibr =
+        new IncrementalBlockReportManager(0, 0, mockMetrics);
+
+    addBlocks(ibr, storage, 0, 5);
+    assertEquals(5, ibr.getPendingBlockCount());
+
+    // Re-adding an existing block (same Block key) must not increase count.
+    ibr.addRDBI(new ReceivedDeletedBlockInfo(
+        new Block(0, 1024, 1000), BlockStatus.DELETED_BLOCK, null), storage);
+    assertEquals(5, ibr.getPendingBlockCount(),
+        "Re-adding an existing block must not change the count");
+
+    ibr.clearIBRs();
+    assertEquals(0, ibr.getPendingBlockCount());
+  }
+
+  /**
+   * The counter must be consistent across multiple storages.
+   */
+  @Test
+  public void testPendingBlockCountMultipleStorages() {
+    IncrementalBlockReportManager ibr =
+        new IncrementalBlockReportManager(0, 0, mockMetrics);
+    DatanodeStorage s1 = new DatanodeStorage("s1");
+    DatanodeStorage s2 = new DatanodeStorage("s2");
+
+    addBlocks(ibr, s1, 0, 5);
+    addBlocks(ibr, s2, 100, 8);
+    assertEquals(13, ibr.getPendingBlockCount());
+  }
+
+  /**
+   * When the pending block count reaches the configured cap,
+   * {@code clearIBRsIfNeeded()} must clear the queue and report that it did.
+   */
+  @Test
+  public void testClearOnSizeCap() {
+    final long cap = 100;
+    IncrementalBlockReportManager ibr =
+        new IncrementalBlockReportManager(0, cap, mockMetrics);
+
+    // Below the cap: nothing should be cleared.
+    addBlocks(ibr, storage, 0, (int) cap - 1);
+    assertFalse(ibr.clearIBRsIfNeeded(),
+        "Queue below cap must not be cleared");
+    assertEquals(cap - 1, ibr.getPendingBlockCount());
+
+    // Reach the cap: the queue must be cleared.
+    addBlocks(ibr, storage, 1000, 1);
+    assertEquals(cap, ibr.getPendingBlockCount());
+    assertTrue(ibr.clearIBRsIfNeeded(),
+        "Queue at cap must be cleared");
+    assertEquals(0, ibr.getPendingBlockCount(),
+        "Queue must be empty after clearing");
+  }
+
+  /**
+   * The cap disabled (0): the queue is never cleared regardless of size,
+   * preserving the historical behavior.
+   */
+  @Test
+  public void testCapDisabled() {
+    IncrementalBlockReportManager ibr =
+        new IncrementalBlockReportManager(0, 0, mockMetrics);
+    addBlocks(ibr, storage, 0, 5000);
+    assertFalse(ibr.clearIBRsIfNeeded(),
+        "With the cap disabled the queue must never be cleared");
+    assertEquals(5000, ibr.getPendingBlockCount());
+  }
+}