From 00252d13a584eca8b259480aff86a37fceb159bd Mon Sep 17 00:00:00 2001 From: singerzhang Date: Thu, 23 Jul 2026 20:51:33 +0800 Subject: [PATCH] HDFS-17953. DataNode IBR pendingIBRs grows unbounded when a NameNode is unreachable, causing OOM --- .../org/apache/hadoop/hdfs/DFSConfigKeys.java | 4 + .../hdfs/server/datanode/BPServiceActor.java | 9 ++ .../hadoop/hdfs/server/datanode/DNConf.java | 4 + .../IncrementalBlockReportManager.java | 66 ++++++++- .../src/main/resources/hdfs-default.xml | 15 ++ .../TestIncrementalBlockReportManager.java | 133 ++++++++++++++++++ .../datanode/TestIncrementalBlockReports.java | 76 ++++++++++ 7 files changed, 303 insertions(+), 4 deletions(-) create mode 100644 hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReportManager.java diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java index 2fb2de5746fafa..209ade618d924f 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java +++ b/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"; diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java index 11006d4698e53e..6267f905ac8046 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java +++ b/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 cmds = null; boolean forceFullBr = scheduler.forceFullBlockReport.getAndSet(false); diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java index 21b92db3073a1f..785824388a395d 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java +++ b/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); diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/IncrementalBlockReportManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/IncrementalBlockReportManager.java index 3b093a7f590130..f6235208531a03 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/IncrementalBlockReportManager.java +++ b/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 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.} 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(); } -} \ No newline at end of file +} diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml index 1c56231fe26c61..9ea67a30cfff2b 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml @@ -4508,6 +4508,21 @@ + + dfs.datanode.ibr.max.pending.blocks + 1000000 + + 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.<nsId> whose RPC address is not configured. + Set to 0 to disable. + + + dfs.checksum.type CRC32C diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReportManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReportManager.java new file mode 100644 index 00000000000000..96e15e29ff374f --- /dev/null +++ b/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()); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReports.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReports.java index 5338ac3035e32d..9f5771f3e91898 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReports.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestIncrementalBlockReports.java @@ -19,6 +19,7 @@ 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.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.atLeastOnce; @@ -525,4 +526,79 @@ public void testIBRRaceCondition3() throws Exception { cluster.shutdown(); } } + + /** + * End-to-end guard against the OOM described in HDFS-17953: when the target + * NameNode is unreachable, the pending IBR queue must not grow without + * bound. With a small {@code dfs.datanode.ibr.max.pending.blocks} cap the + * queue must stay bounded even though every IBR send fails and would + * otherwise be re-queued forever, and a full block report must be scheduled + * so the NameNode can resync once it is reachable again. + */ + @Test + @Timeout(value = 120) + public void testIBRQueueBoundedWhenNNUnreachable() throws Exception { + // Rebuild the cluster with a small IBR cap so we can reach it quickly. + cluster.shutdown(); + cluster = null; + final int cap = 20; + Configuration capConf = new HdfsConfiguration(); + capConf.setLong(DFSConfigKeys.DFS_DATANODE_IBR_MAX_PENDING_BLOCKS_KEY, cap); + // Disable automatic IBR sending timer effects by keeping default interval; + // we drive sends explicitly via the actor guard below. + cluster = new MiniDFSCluster.Builder(capConf).numDataNodes(1).build(); + try { + cluster.waitActive(); + singletonNn = cluster.getNameNode(); + singletonDn = cluster.getDataNodes().get(0); + bpos = singletonDn.getAllBpOs().get(0); + actor = bpos.getBPServiceActors().get(0); + try (FsDatasetSpi.FsVolumeReferences volumes = + singletonDn.getFSDataset().getFsVolumeReferences()) { + storageUuid = volumes.get(0).getStorageID(); + } + + // Make the NN appear unreachable: every IBR RPC throws. + DatanodeProtocolClientSideTranslatorPB nnSpy = spyOnDnCallsToNn(); + doAnswer((InvocationOnMock inv) -> { + throw new IOException("Simulated NameNode unreachable"); + }).when(nnSpy).blockReceivedAndDeleted( + any(DatanodeRegistration.class), + anyString(), + any(StorageReceivedDeletedBlocks[].class)); + + IncrementalBlockReportManager ibr = actor.getIbrManager(); + DatanodeStorage s = singletonDn.getFSDataset().getStorage(storageUuid); + + // Inject many more distinct blocks than the cap. Without the fix the + // queue would grow to 10*cap; with the fix the periodic guard keeps it + // bounded by the cap. + long observedMax = 0; + for (int i = 0; i < cap * 10; i++) { + ibr.addRDBI(new ReceivedDeletedBlockInfo( + new Block(10000 + i, 1024, 2000 + i), + BlockStatus.DELETED_BLOCK, null), s); + // Emulate the guard that BPServiceActor.offerService runs each cycle. + ibr.clearIBRsIfNeeded(); + observedMax = Math.max(observedMax, ibr.getPendingBlockCount()); + } + + assertTrue(observedMax <= cap, + "Pending IBR block count must never exceed the cap; observed " + + observedMax + ", cap " + cap); + + // A full block report must have been scheduled after a clear so the NN + // can resync once reachable. forceFullBlockReportNow sets this flag. + // (The actor sets it via scheduler.forceFullBlockReportNow(); here we + // assert the manager cleared at least once by checking the count is + // strictly below the total injected.) + assertTrue(ibr.getPendingBlockCount() < cap * 10, + "Queue must have been cleared at least once"); + } finally { + if (cluster != null) { + cluster.shutdown(); + cluster = null; + } + } + } }