Copilot commented on code in PR #10122:
URL: https://github.com/apache/ozone/pull/10122#discussion_r3548252913
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -422,6 +435,57 @@ 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 inflightReconstructionCount.get();
+ }
+
+ /**
+ * 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() {
+ int limit = getReconstructionInFlightLimit();
+ return limit > 0 && getInflightReconstructionCount() >= limit;
+ }
+
+ private void decrementInflightReconstructionCount() {
+ inflightReconstructionCount.updateAndGet(count -> Math.max(0, count - 1));
+ }
+
+ private boolean tryReserveReconstructionSlot() {
+ int limit = getReconstructionInFlightLimit();
+ if (limit <= 0) {
+ return true;
+ }
+ while (true) {
+ int current = inflightReconstructionCount.get();
+ if (current >= limit) {
+ return false;
+ }
+ if (inflightReconstructionCount.compareAndSet(current, current + 1)) {
+ return true;
+ }
+ }
+ }
Review Comment:
tryReserveReconstructionSlot() returns true without incrementing
inflightReconstructionCount when the limit is disabled (<= 0). That makes
getInflightReconstructionCount() always return 0 in the default configuration,
contradicting its Javadoc (“active … currently in progress”) and undermining
any future use of this counter for observability.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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);
Review Comment:
This line is much longer than nearby constructor argument formatting in this
test (eg cmd1/cmd2 in testInflightReconstructionLimit). Please wrap the final
arguments onto a new line for readability and to match existing style in this
file.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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);
Review Comment:
This line is much longer than nearby constructor argument formatting in this
test (eg cmd1/cmd2 in testInflightReconstructionLimit). Please wrap the final
arguments onto a new line for readability and to match existing style in this
file.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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);
Review Comment:
This line is much longer than nearby constructor argument formatting in this
test (eg cmd1/cmd2 in testInflightReconstructionLimit). Please wrap the final
arguments onto a new line for readability and to match existing style in this
file.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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(1.1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setEcDecommissionReconstructionLoadFactor(0.9);
+ config.setReconstructionGlobalLimit(-1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setReconstructionGlobalLimit(0);
+ 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 containerId = t + 100;
+ executor.submit(() -> {
+ try {
+ startLatch.await();
+ for (int i = 0; i < attemptsPerThread; i++) {
+ ReconstructECContainersCommand cmd = new
ReconstructECContainersCommand(
+ containerId + i, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+
ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
+ (ECReplicationConfig) repConfig);
Review Comment:
The variable name `containerId` is misleading here: it is used as the
command ID base passed into ReconstructECContainersCommand, not a container ID
(the ContainerInfo is constant). Renaming avoids confusion when reading the
concurrency test.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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);
Review Comment:
This line is much longer than nearby constructor argument formatting in this
test (eg cmd1/cmd2 in testInflightReconstructionLimit). Please wrap the final
arguments onto a new line for readability and to match existing style in this
file.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -537,17 +601,31 @@ public void sendThrottledReplicationCommand(ContainerInfo
containerInfo,
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()) {
+ if (!tryReserveReconstructionSlot()) {
metrics.incrECReconstructionCmdsDeferredTotal();
- throw new CommandTargetOverloadedException("No target with capacity " +
- "available for reconstruction of " + containerInfo.getContainerID());
+ throw new CommandTargetOverloadedException(
+ "Global reconstruction limit (" + getReconstructionInFlightLimit()
+ + ") reached for container " + containerInfo.getContainerID());
Review Comment:
The PR description says UnhealthyReplicationProcessor checks the global
reconstruction limit before dequeuing, but the only enforcement here is via
CommandTargetOverloadedException in sendThrottledReconstructionCommand().
UnhealthyReplicationProcessor currently catches
CommandTargetOverloadedException and continues dequeuing/requeueing, so hitting
the global limit will likely cause churn instead of ending the iteration early
(similar to the existing inflight replication limit behavior).
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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);
Review Comment:
This line is much longer than nearby constructor argument formatting in this
test (eg cmd1/cmd2 in testInflightReconstructionLimit). Please wrap the final
arguments onto a new line for readability and to match existing style in this
file.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ 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()),
+ 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(1.1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setEcDecommissionReconstructionLoadFactor(0.9);
+ config.setReconstructionGlobalLimit(-1);
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config.setReconstructionGlobalLimit(0);
+ 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 containerId = t + 100;
+ executor.submit(() -> {
+ try {
+ startLatch.await();
+ for (int i = 0; i < attemptsPerThread; i++) {
+ ReconstructECContainersCommand cmd = new
ReconstructECContainersCommand(
+ containerId + 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();
+ }
+ });
+ }
+
+ startLatch.countDown();
+ assertTrue(doneLatch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+
Review Comment:
If doneLatch.await(...) times out, the assert will throw and
executor.shutdown() will be skipped, potentially leaking threads into the rest
of the test suite. Put the shutdown into a finally block so it always executes.
--
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]
