Copilot commented on code in PR #10122:
URL: https://github.com/apache/ozone/pull/10122#discussion_r3548006705
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -1291,6 +1352,39 @@ 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 = "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."
+ )
+ 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 = "The threshold factor (between 0 and 1) of a node's " +
+ "replication limit at which SCM switches to reconstruction for " +
+ "EC decommission. Default is 0.9."
+ )
+ private double ecDecommissionReconstructionLoadFactor = 0.9;
Review Comment:
The config description states ecDecommissionReconstructionLoadFactor is
"between 0 and 1", but ReplicationManagerConfiguration.validate() does not
enforce this (nor does it validate reconstructionGlobalLimit >= 0). Without
validation, invalid values can silently disable or distort the intended
behavior.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -1074,6 +1122,19 @@ 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) {
+ long cmdId = op.getCommand().getId();
+ reconstructionCommandIdToPendingFragmentCount.compute(cmdId, (k, v) -> {
+ if (v == null || v <= 1) {
+ inflightReconstructionCount.decrementAndGet();
+ return null;
+ }
+ return v - 1;
+ });
Review Comment:
In opCompleted, the reconstruction inflight counter is decremented even when
the command ID is not present in reconstructionCommandIdToPendingFragmentCount
(v == null). This can drive inflightReconstructionCount negative (eg due to
duplicate completion notifications or after state resets), which would
incorrectly disable throttling and corrupt metrics. Only decrement when there
is a tracked entry.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestUnderReplicatedProcessor.java:
##########
@@ -131,4 +131,21 @@ public void testMessageNotProcessedIfGlobalLimitReached()
throws IOException {
assertEquals(1, rmMetrics.getPendingReplicationLimitReachedTotal());
}
+ @Test
+ public void testProcessorContinuesWhenReconstructionLimitReached()
+ throws IOException {
+ when(replicationManager.isReconstructionLimitReached()).thenReturn(true);
+
when(replicationManager.processUnderReplicatedContainer(any())).thenReturn(1);
Review Comment:
This test stubs replicationManager.isReconstructionLimitReached(), but
UnderReplicatedProcessor / UnhealthyReplicationProcessor never calls that
method, so the stub is unused and the test does not actually cover
reconstruction-limit behavior. Either remove the unused stub, or add assertions
against code that consults the reconstruction limit.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1803,136 @@ private void mockReplicationCommandCounts(
});
}
+ @Test
+ public void testReconstructionGlobalLimitDisabledByDefault() throws
IOException {
+ assertEquals(0, rmConf.getReconstructionGlobalLimit());
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(0, rm.getReconstructionInFlightLimit());
+ 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()),
+ 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()),
+ 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()),
+ integers2ByteString(ImmutableList.of(1)), (ECReplicationConfig)
repConfig);
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ integers2ByteString(ImmutableList.of(2)), (ECReplicationConfig)
repConfig);
+ ReconstructECContainersCommand cmd3 = new ReconstructECContainersCommand(
+ 3L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ 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()),
+ 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());
+ }
+
+ private static ByteString integers2ByteString(List<Integer> src) {
+ byte[] dst = new byte[src.size()];
+ for (int i = 0; i < src.size(); i++) {
+ dst[i] = src.get(i).byteValue();
+ }
+ return dst.length > 0 ? UnsafeByteOperations.unsafeWrap(dst)
+ : ByteString.EMPTY;
+ }
Review Comment:
This helper duplicates ECUnderReplicationHandler.integers2ByteString with
identical logic. Delegating to the existing helper avoids drift and keeps
encoding consistent across tests and production code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
