This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 78784ba9fac HDDS-15071. [SCM] Add configuration and global EC
reconstruction limit (#11054)
78784ba9fac is described below
commit 78784ba9fac6db9051a43f3bfa2a125bbc016b39
Author: Wei-Chiu Chuang <[email protected]>
AuthorDate: Thu Oct 8 15:59:16 2026 -0600
HDDS-15071. [SCM] Add configuration and global EC reconstruction limit
(#11054)
Generated-by: Cursor <[email protected]>
---
.../container/replication/ReplicationManager.java | 296 +++++++++++++-
.../replication/TestReplicationManager.java | 440 +++++++++++++++++++++
2 files changed, 716 insertions(+), 20 deletions(-)
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
index 1742f735cc9..29f00a10d3c 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
@@ -37,6 +37,7 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -158,12 +159,15 @@ public class ReplicationManager implements SCMService,
ContainerReplicaPendingOp
private final Map<DatanodeDetails, Integer> excludedNodes =
new ConcurrentHashMap<>();
+ private final GlobalReconstructionTracker globalReconstructionTracker =
+ new GlobalReconstructionTracker();
+
/**
* SCMService related variables.
* After leaving safe mode, replicationMonitor needs to wait for a while
* before really take effect.
*/
- private final Lock serviceLock = new ReentrantLock();
+ private final Lock serviceLifecycleLock = new ReentrantLock();
private ServiceStatus serviceStatus = ServiceStatus.PAUSING;
private final long waitTimeInMillis;
private long lastTimeToBeReadyInMillis = 0;
@@ -421,6 +425,36 @@ public long getInflightReplicationCount() {
.getPendingOpCount(ContainerReplicaOp.PendingOpType.ADD);
}
+ /**
+ * Returns the number of active EC reconstruction commands currently in
+ * progress across the cluster.
+ */
+ public int getInflightReconstructionCount() {
+ return globalReconstructionTracker.getInflightCount();
+ }
+
+ int getReconstructionPendingFragmentCount(long cmdId) {
+ return globalReconstructionTracker.getPendingFragmentCount(cmdId);
+ }
+
+ /**
+ * Returns the maximum number of inflight reconstruction commands allowed
+ * across the cluster at any given time.
+ * @return the maximum number of inflight reconstruction commands allowed
+ */
+ public int getReconstructionInFlightLimit() {
+ return rmConf.getReconstructionGlobalLimit();
+ }
+
+ /**
+ * Returns true if the number of inflight reconstruction commands has reached
+ * the global limit.
+ * @return true if the limit is reached, false otherwise
+ */
+ public boolean isReconstructionLimitReached() {
+ return globalReconstructionTracker.isLimitReached();
+ }
+
/**
* Sends delete container command for the given container to the given
* datanode.
@@ -533,20 +567,45 @@ public void sendThrottledReplicationCommand(ContainerInfo
containerInfo,
sendDatanodeCommand(cmd, containerInfo, source);
}
+ /**
+ * Send a {@link ReconstructECContainersCommand} to a datanode with capacity,
+ * after reserving a cluster-wide reconstruction slot.
+ *
+ * <p>The command may encode multiple missing EC indexes (fragments). One
+ * global slot is reserved per command; {@link #opCompleted} releases it
+ * incrementally as each fragment's pending ADD op completes or expires.
+ *
+ * <p>If the global reconstruction limit is reached, or no target datanode
+ * has spare replication capacity, a {@link CommandTargetOverloadedException}
+ * is thrown and the command is not sent. Callers (e.g.
+ * {@code ECUnderReplicationHandler}) can fall back to 1-1 replication.
+ *
+ * @param containerInfo the container to reconstruct
+ * @param command the reconstruction command to send
+ */
public void sendThrottledReconstructionCommand(ContainerInfo containerInfo,
ReconstructECContainersCommand command)
throws CommandTargetOverloadedException, NotLeaderException {
- List<DatanodeDetails> targets = command.getTargetDatanodes();
- List<Pair<Integer, DatanodeDetails>> targetWithCmds =
- getAvailableDatanodesForReplication(targets);
- if (targetWithCmds.isEmpty()) {
- metrics.incrECReconstructionCmdsDeferredTotal();
- throw new CommandTargetOverloadedException("No target with capacity " +
- "available for reconstruction of " + containerInfo.getContainerID());
- }
- DatanodeDetails target = selectAndOptionallyExcludeDatanode(
- rmConf.getReconstructionCommandWeight(), targetWithCmds);
- sendDatanodeCommand(command, containerInfo, target);
+ globalReconstructionTracker.reserveAndSend(
+ command.getId(),
+ command.getMissingContainerIndexes().size(),
+ containerInfo.getContainerID(),
+ () -> {
+ // Pick the least-loaded target datanode that can accept the command.
+ List<DatanodeDetails> targets = command.getTargetDatanodes();
+ List<Pair<Integer, DatanodeDetails>> targetWithCmds =
+ getAvailableDatanodesForReplication(targets);
+ if (targetWithCmds.isEmpty()) {
+ metrics.incrECReconstructionCmdsDeferredTotal();
+ throw new CommandTargetOverloadedException("No target with
capacity " +
+ "available for reconstruction of " +
containerInfo.getContainerID());
+ }
+ DatanodeDetails target = selectAndOptionallyExcludeDatanode(
+ rmConf.getReconstructionCommandWeight(), targetWithCmds);
+ // sendDatanodeCommand schedules one pending ADD per missing index
and
+ // increments ecReconstructionCmdsSentTotal.
+ sendDatanodeCommand(command, containerInfo, target);
+ });
}
private DatanodeDetails selectAndOptionallyExcludeDatanode(
@@ -1055,6 +1114,12 @@ ReplicationQueue getQueue() {
@Override
public void opCompleted(ContainerReplicaOp op, ContainerID containerID,
boolean timedOut) {
+ if (op.getOpType() == ContainerReplicaOp.PendingOpType.ADD
+ && op.getCommand() != null
+ && op.getCommand().getType() == Type.reconstructECContainersCommand) {
+ globalReconstructionTracker.onFragmentComplete(op.getCommand().getId());
+ }
+
if (!(timedOut && op.getOpType() ==
ContainerReplicaOp.PendingOpType.DELETE)) {
// We only care about expired delete ops. All others should be ignored.
return;
@@ -1262,6 +1327,41 @@ public static class ReplicationManagerConfiguration
)
private int containerSampleLimit = 100;
+ @Config(key =
"hdds.scm.replication.decommission.ec.reconstruction.enabled",
+ type = ConfigType.BOOLEAN,
+ defaultValue = "false",
+ reconfigurable = true,
+ tags = { SCM },
+ description = "Reserved for HDDS-15072. When implemented, if true, SCM
" +
+ "will switch from 1-1 replication to multi-source reconstruction
for " +
+ "EC containers on decommissioning nodes when the node's load
exceeds " +
+ "the threshold. Not read by SCM in this release."
+ )
+ private boolean ecDecommissionReconstructionEnabled = false;
+
+ @Config(key =
"hdds.scm.replication.decommission.ec.reconstruction.load.factor",
+ type = ConfigType.DOUBLE,
+ defaultValue = "0.9",
+ reconfigurable = true,
+ tags = { SCM },
+ description = "Reserved for HDDS-15072. When implemented, the
threshold " +
+ "factor (between 0 and 1) of a node's replication limit at which "
+
+ "SCM switches to reconstruction for EC decommission. Not read by "
+
+ "SCM in this release. Default is 0.9."
+ )
+ private double ecDecommissionReconstructionLoadFactor = 0.9;
+
+ @Config(key = "hdds.scm.replication.reconstruction.global.limit",
+ type = ConfigType.INT,
+ defaultValue = "0",
+ reconfigurable = true,
+ tags = { SCM },
+ description = "A cluster-wide limit to restrict the total number of " +
+ "active EC reconstruction commands across the cluster. A setting "
+
+ "of zero disables global reconstruction limit checking."
+ )
+ private int reconstructionGlobalLimit = 0;
+
@Config(key = "hdds.scm.replication.quasi.closed.stuck.best.origin.copies",
type = ConfigType.INT,
defaultValue = "3",
@@ -1318,6 +1418,30 @@ public void setDatanodeReplicationLimit(int limit) {
this.datanodeReplicationLimit = limit;
}
+ public boolean isEcDecommissionReconstructionEnabled() {
+ return ecDecommissionReconstructionEnabled;
+ }
+
+ public void setEcDecommissionReconstructionEnabled(boolean enabled) {
+ this.ecDecommissionReconstructionEnabled = enabled;
+ }
+
+ public double getEcDecommissionReconstructionLoadFactor() {
+ return ecDecommissionReconstructionLoadFactor;
+ }
+
+ public void setEcDecommissionReconstructionLoadFactor(double factor) {
+ this.ecDecommissionReconstructionLoadFactor = factor;
+ }
+
+ public int getReconstructionGlobalLimit() {
+ return reconstructionGlobalLimit;
+ }
+
+ public void setReconstructionGlobalLimit(int limit) {
+ this.reconstructionGlobalLimit = limit;
+ }
+
public void setMaintenanceRemainingRedundancy(int redundancy) {
this.maintenanceRemainingRedundancy = redundancy;
}
@@ -1411,7 +1535,8 @@ public void validate() {
+ " must be >= datanode.reconstruction.weight: "
+ reconstructionCommandWeight);
}
- if (inflightReplicationLimitFactor < 0) {
+ if (!Double.isFinite(inflightReplicationLimitFactor)
+ || inflightReplicationLimitFactor < 0) {
throw new IllegalArgumentException(
"inflight.limit.factor is set to " + inflightReplicationLimitFactor
+ " and must be >= 0");
@@ -1421,12 +1546,28 @@ public void validate() {
"inflight.limit.factor is set to " + inflightReplicationLimitFactor
+ " and must be <= 1");
}
+ if (!Double.isFinite(ecDecommissionReconstructionLoadFactor)
+ || ecDecommissionReconstructionLoadFactor < 0) {
+ throw new IllegalArgumentException(
+ "decommission.ec.reconstruction.load.factor is set to "
+ + ecDecommissionReconstructionLoadFactor + " and must be >=
0");
+ }
+ if (ecDecommissionReconstructionLoadFactor > 1) {
+ throw new IllegalArgumentException(
+ "decommission.ec.reconstruction.load.factor is set to "
+ + ecDecommissionReconstructionLoadFactor + " and must be <=
1");
+ }
+ if (reconstructionGlobalLimit < 0) {
+ throw new IllegalArgumentException(
+ "reconstruction.global.limit is set to " +
reconstructionGlobalLimit
+ + " and must be >= 0");
+ }
}
}
@Override
public void notifyStatusChanged() {
- serviceLock.lock();
+ serviceLifecycleLock.lock();
try {
// 1) SCMContext#isLeaderReady returns true.
// 2) not in safe mode.
@@ -1439,28 +1580,28 @@ public void notifyStatusChanged() {
// and then back to a leader in a short time, there may be old
pending
// Ops in the ContainerReplicaPendingOps table. They are no longer
// needed as the DN will discard any commands when the term changes.
- // Therefore we should clear the table so RM starts from a clean
- // state.
- containerReplicaPendingOps.clear();
+ // Reset pending ops and reconstruction tracking atomically under the
+ // tracker lock (see GlobalReconstructionTracker#clear).
+ globalReconstructionTracker.clear();
serviceStatus = ServiceStatus.RUNNING;
}
} else {
serviceStatus = ServiceStatus.PAUSING;
}
} finally {
- serviceLock.unlock();
+ serviceLifecycleLock.unlock();
}
}
@Override
public boolean shouldRun() {
- serviceLock.lock();
+ serviceLifecycleLock.lock();
try {
// If safe mode is off, then this SCMService starts to run with a delay.
return serviceStatus == ServiceStatus.RUNNING &&
clock.millis() - lastTimeToBeReadyInMillis >= waitTimeInMillis;
} finally {
- serviceLock.unlock();
+ serviceLifecycleLock.unlock();
}
}
@@ -1586,5 +1727,120 @@ public synchronized boolean notifyNodeStateChange() {
return false;
}
}
+
+ /**
+ * Cluster-wide EC reconstruction slot accounting.
+ *
+ * <p>All mutations to the inflight reconstruction counter and per-command
+ * fragment map go through this class. External callers do not acquire locks.
+ *
+ * <p>Lock policy:
+ * <ul>
+ * <li>{@link #reserveAndSend} holds {@link #lock} across reserve, map
+ * registration, and send so {@link #clear()} on leader transition
+ * cannot run mid-send.</li>
+ * <li>{@link #clear()} clears {@link ContainerReplicaPendingOps} under
+ * the same lock so pending ADD ops cannot be scheduled after the
+ * pending-op table is reset.</li>
+ * <li>{@link #onFragmentComplete} runs without {@link #lock}; it uses
+ * {@code ConcurrentHashMap.compute} and tolerates a prior
+ * {@link #clear()}.</li>
+ * </ul>
+ */
+ private final class GlobalReconstructionTracker {
+
+ private final Lock lock = new ReentrantLock();
+ private final AtomicInteger inflightCount = new AtomicInteger(0);
+ private final Map<Long, Integer> pendingFragmentCountByCommandId =
+ new ConcurrentHashMap<>();
+
+ int getInflightCount() {
+ return inflightCount.get();
+ }
+
+ int getPendingFragmentCount(long cmdId) {
+ return pendingFragmentCountByCommandId.getOrDefault(cmdId, 0);
+ }
+
+ boolean isLimitReached() {
+ int limit = getReconstructionInFlightLimit();
+ return limit > 0 && getInflightCount() >= limit;
+ }
+
+ void reserveAndSend(long cmdId, int fragmentCount, long containerId,
+ ReconstructionSendAction send)
+ throws CommandTargetOverloadedException, NotLeaderException {
+ if (fragmentCount <= 0) {
+ throw new IllegalArgumentException(
+ "Reconstruction command " + cmdId + " for container " + containerId
+ + " has no fragments");
+ }
+ lock.lock();
+ try {
+ if (!tryReserve()) {
+ metrics.incrECReconstructionCmdsDeferredTotal();
+ throw new CommandTargetOverloadedException(
+ "Global reconstruction limit (" +
getReconstructionInFlightLimit()
+ + ") reached for container " + containerId);
+ }
+ pendingFragmentCountByCommandId.put(cmdId, fragmentCount);
+ boolean sent = false;
+ try {
+ send.run();
+ sent = true;
+ } finally {
+ if (!sent) {
+ pendingFragmentCountByCommandId.remove(cmdId);
+ releaseSlot();
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ void onFragmentComplete(long cmdId) {
+ pendingFragmentCountByCommandId.compute(cmdId, (k, pendingFragments) -> {
+ if (pendingFragments == null) {
+ return null;
+ }
+ if (pendingFragments <= 1) {
+ releaseSlot();
+ return null;
+ }
+ return pendingFragments - 1;
+ });
+ }
+
+ void clear() {
+ lock.lock();
+ try {
+ containerReplicaPendingOps.clear();
+ pendingFragmentCountByCommandId.clear();
+ inflightCount.set(0);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private boolean tryReserve() {
+ int limit = getReconstructionInFlightLimit();
+ int current = inflightCount.get();
+ if (limit > 0 && current >= limit) {
+ return false;
+ }
+ inflightCount.incrementAndGet();
+ return true;
+ }
+
+ private void releaseSlot() {
+ inflightCount.updateAndGet(count -> Math.max(0, count - 1));
+ }
+ }
+
+ @FunctionalInterface
+ private interface ReconstructionSendAction {
+ void run() throws CommandTargetOverloadedException, NotLeaderException;
+ }
}
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
index bcb3dea9768..032789d180a 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
@@ -44,10 +44,12 @@
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.isNull;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.protobuf.UnsafeByteOperations;
import java.io.IOException;
@@ -61,8 +63,13 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -1739,4 +1746,437 @@ private void mockReplicationCommandCounts(
});
}
+ @Test
+ public void testReconstructionGlobalLimitDisabledByDefault()
+ throws IOException, NodeNotFoundException {
+ assertEquals(0, rmConf.getReconstructionGlobalLimit());
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(0, rm.getReconstructionInFlightLimit());
+ assertFalse(rm.isReconstructionLimitReached());
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+ }
+
+ @Test
+ public void testInflightReconstructionLimit() throws IOException,
NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(2, rm.getReconstructionInFlightLimit());
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ // Send one reconstruction command with 2 fragments
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd1);
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Send another reconstruction command with 1 fragment
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd2);
+ assertEquals(2, rm.getInflightReconstructionCount());
+ assertTrue(rm.isReconstructionLimitReached());
+
+ // Complete one fragment of cmd1
+ ContainerReplicaOp op1 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(0), 1, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op1, container.containerID(), false);
+ // Still 2 because cmd1 is not fully finished
+ assertEquals(2, rm.getInflightReconstructionCount());
+
+ // Complete second fragment of cmd1
+ ContainerReplicaOp op2 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(1), 2, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op2, container.containerID(), false);
+ // Now 1
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Complete cmd2
+ ContainerReplicaOp op3 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd2.getTargetDatanodes().get(0), 3, cmd2, Long.MAX_VALUE, 0);
+ rm.opCompleted(op3, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testSendReconstructionCommandRejectedWhenGlobalLimitReached()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
+ (ECReplicationConfig) repConfig);
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(2)),
+ (ECReplicationConfig) repConfig);
+ ReconstructECContainersCommand cmd3 = new ReconstructECContainersCommand(
+ 3L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd1);
+ rm.sendThrottledReconstructionCommand(container, cmd2);
+ assertEquals(2, rm.getInflightReconstructionCount());
+
+ assertThrows(CommandTargetOverloadedException.class,
+ () -> rm.sendThrottledReconstructionCommand(container, cmd3));
+ assertEquals(2, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testNotifyStatusChangedClearsReconstructionCounters()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+ enableProcessAll();
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ assertEquals(1, rm.getInflightReconstructionCount());
+
+ when(scmContext.isLeaderReady()).thenReturn(false);
+ rm.notifyStatusChanged();
+ when(scmContext.isLeaderReady()).thenReturn(true);
+ when(scmContext.isInSafeMode()).thenReturn(false);
+ rm.notifyStatusChanged();
+
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testInflightReconstructionCountNotNegativeAfterFailoverClear()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+ enableProcessAll();
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ assertEquals(1, rm.getInflightReconstructionCount());
+
+ when(scmContext.isLeaderReady()).thenReturn(false);
+ rm.notifyStatusChanged();
+ when(scmContext.isLeaderReady()).thenReturn(true);
+ when(scmContext.isInSafeMode()).thenReturn(false);
+ rm.notifyStatusChanged();
+ assertEquals(0, rm.getInflightReconstructionCount());
+
+ ContainerReplicaOp op1 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd.getTargetDatanodes().get(0), 1, cmd, Long.MAX_VALUE, 0);
+ ContainerReplicaOp op2 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd.getTargetDatanodes().get(1), 2, cmd, Long.MAX_VALUE, 0);
+
+ rm.opCompleted(op1, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ rm.opCompleted(op2, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ rm.opCompleted(op1, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testReconstructionConfigValidation() {
+ ReplicationManager.ReplicationManagerConfiguration config =
+ new ReplicationManager.ReplicationManagerConfiguration();
+
+ config.setEcDecommissionReconstructionLoadFactor(-0.1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setEcDecommissionReconstructionLoadFactor(Double.NaN);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setEcDecommissionReconstructionLoadFactor(1.1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setEcDecommissionReconstructionLoadFactor(0.9);
+ config.setReconstructionGlobalLimit(-1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setReconstructionGlobalLimit(0);
+ config.setInflightReplicationLimitFactor(Double.NaN);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setInflightReplicationLimitFactor(0.75);
+ config.validate();
+ }
+
+ @Test
+ public void testReconstructionGlobalLimitEnforcedConcurrently()
+ throws InterruptedException, NodeNotFoundException, IOException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ int threadCount = 8;
+ int attemptsPerThread = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ CountDownLatch doneLatch = new CountDownLatch(threadCount);
+ AtomicInteger accepted = new AtomicInteger();
+ AtomicInteger rejected = new AtomicInteger();
+
+ for (int t = 0; t < threadCount; t++) {
+ final long commandIdBase = t + 100;
+ executor.submit(() -> {
+ try {
+ startLatch.await();
+ for (int i = 0; i < attemptsPerThread; i++) {
+ ReconstructECContainersCommand cmd = new
ReconstructECContainersCommand(
+ commandIdBase + i, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+
ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
+ (ECReplicationConfig) repConfig);
+ try {
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ accepted.incrementAndGet();
+ } catch (CommandTargetOverloadedException | NotLeaderException e) {
+ rejected.incrementAndGet();
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ }
+
+ try {
+ startLatch.countDown();
+ assertTrue(doneLatch.await(30, TimeUnit.SECONDS));
+ assertEquals(2, rm.getInflightReconstructionCount());
+ assertEquals(2, accepted.get());
+ assertTrue(rejected.get() > 0);
+ } finally {
+ executor.shutdown();
+ }
+ }
+
+ @Test
+ public void testReconstructionMapRegisteredBeforePendingOpsScheduled()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ assertEquals(2, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ assertEquals(1, rm.getInflightReconstructionCount());
+
+ ContainerReplicaOp op = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd.getTargetDatanodes().get(0), 1, cmd, Long.MAX_VALUE, 0);
+ rm.opCompleted(op, container.containerID(), false);
+ assertEquals(1, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ assertEquals(1, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testInflightReconstructionCountDecrementsWhenAddOpExpires()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ assertEquals(2, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ assertEquals(1, rm.getInflightReconstructionCount());
+
+ ContainerReplicaOp firstAdd = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd.getTargetDatanodes().get(0), 1, cmd, Long.MAX_VALUE, 0);
+ rm.opCompleted(firstAdd, container.containerID(), true);
+ assertEquals(1, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ assertEquals(1, rm.getInflightReconstructionCount());
+
+ ContainerReplicaOp secondAdd = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd.getTargetDatanodes().get(1), 2, cmd, Long.MAX_VALUE, 0);
+ rm.opCompleted(secondAdd, container.containerID(), true);
+ assertEquals(0, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testReconstructionWithNoFragmentsDoesNotReserveSlot()
+ throws NodeNotFoundException, IOException {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ Collections.emptyList(),
+ ECUnderReplicationHandler.integers2ByteString(Collections.emptyList()),
+ (ECReplicationConfig) repConfig);
+
+ assertThrows(IllegalArgumentException.class,
+ () -> rm.sendThrottledReconstructionCommand(container, cmd));
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertEquals(0, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ verify(nodeManager, never()).addDatanodeCommand(any(), any());
+ }
+
+ @Test
+ public void testReconstructionSlotReleasedAndMapClearedOnSendFailure()
+ throws NodeNotFoundException, IOException {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+
+ int limit = rm.getConfig().getDatanodeReplicationLimit();
+ int reconstructionWeight = rm.getConfig().getReconstructionCommandWeight();
+ int reconstructionCount = 2;
+ int replicationCount = limit - reconstructionCount * reconstructionWeight;
+ mockReplicationCommandCounts(any -> replicationCount, any ->
reconstructionCount);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = createReconstructionCommand(
+ container, MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails());
+
+ assertThrows(CommandTargetOverloadedException.class,
+ () -> rm.sendThrottledReconstructionCommand(container, cmd));
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertEquals(0, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+ }
+
+ @Test
+ public void testNotifyStatusChangedBlocksUntilReconstructionSendCompletes()
+ throws Exception {
+ rmConf.setReconstructionGlobalLimit(10);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+ enableProcessAll();
+
+ CountDownLatch sendReachedAddCommand = new CountDownLatch(1);
+ CountDownLatch releaseSend = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ sendReachedAddCommand.countDown();
+ try {
+ releaseSend.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ return null;
+ }).when(nodeManager).addDatanodeCommand(any(), any());
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+ ReconstructECContainersCommand cmd = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
+ (ECReplicationConfig) repConfig);
+
+ Thread sender = new Thread(() -> {
+ try {
+ rm.sendThrottledReconstructionCommand(container, cmd);
+ } catch (CommandTargetOverloadedException | NotLeaderException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ sender.start();
+ assertTrue(sendReachedAddCommand.await(30, TimeUnit.SECONDS));
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertEquals(1, rm.getReconstructionPendingFragmentCount(cmd.getId()));
+
+ AtomicBoolean notifyCompleted = new AtomicBoolean(false);
+ Thread notifyThread = new Thread(() -> {
+ rm.notifyStatusChanged();
+ notifyCompleted.set(true);
+ });
+ notifyThread.start();
+ notifyThread.join(500);
+ assertFalse(notifyCompleted.get(),
+ "notifyStatusChanged must wait for in-flight reconstruction send");
+ assertEquals(1, rm.getInflightReconstructionCount());
+
+ releaseSend.countDown();
+ sender.join(TimeUnit.SECONDS.toMillis(30));
+ notifyThread.join(TimeUnit.SECONDS.toMillis(30));
+ assertTrue(notifyCompleted.get());
+
assertTrue(containerReplicaPendingOps.getPendingOps(container.containerID())
+ .isEmpty());
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
