jojochuang commented on code in PR #10122:
URL: https://github.com/apache/ozone/pull/10122#discussion_r3599328618
##########
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:
Fixed in aad8da6d69d: wrapped long constructor arguments onto separate lines.
##########
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:
Fixed in aad8da6d69d: wrapped long constructor arguments onto separate lines.
##########
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:
Fixed in aad8da6d69d: moved `executor.shutdown()` into a `finally` block.
##########
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:
Fixed in aad8da6d69d: renamed `containerId` to `commandIdBase` in the
concurrent test.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -421,6 +434,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;
+ }
Review Comment:
Fixed in aad8da6d69d: counter is always incremented on reserve; limit
enforcement only applies when `reconstruction.global.limit > 0`, so runtime
reconfig sees an accurate inflight count.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -421,6 +434,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;
+ }
Review Comment:
PR description updated to document send-path enforcement in
`sendThrottledReconstructionCommand()` rather than processor dequeue.
--
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]
