smengcl commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3692593385
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java:
##########
@@ -1052,4 +1070,449 @@ private void scheduleTasks(
rs.addTask(new ReplicationTask(toTarget(i, target), noopReplicator));
}
}
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumeDisabledUsesGlobalPool(ContainerLayoutVersion layout) {
+ this.layoutVersion = layout;
+ ReplicationServer.ReplicationConfig repConf =
+ new ReplicationServer.ReplicationConfig();
+ repConf.setPerVolumeEnabled(false);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ assertNull(supervisor.getVolumeReplicationThreadPools());
+ replicatorRef.set(doneReplicator);
+ supervisor.addTask(createTask(1L));
+ assertEquals(1, supervisor.getReplicationSuccessCount());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumeInitLogging(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ LogCapturer supervisorLogs =
+ LogCapturer.captureLogs(ReplicationSupervisor.class);
+ LogCapturer poolLogs =
+ LogCapturer.captureLogs(VolumeReplicationThreadPools.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ assertNotNull(supervisor.getVolumeReplicationThreadPools());
+ assertThat(supervisorLogs.getOutput())
+ .contains("Per-volume container replication thread pools enabled");
+ assertThat(poolLogs.getOutput())
+ .contains("Initialized 2 per-volume replication thread pools");
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertThat(poolLogs.getOutput())
+ .contains(volume.getStorageDir().getPath());
+ }
+ } finally {
+ supervisorLogs.stopCapturing();
+ poolLogs.stopCapturing();
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolSizeRespected(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 3);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(3, pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolResize(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ supervisor.setPerVolumePoolSize(3);
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(3, pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ assertEquals(3, repConf.getPerVolumeStreamsLimit());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolResizeOnNodeStateChange(ContainerLayoutVersion
layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 2);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ try {
+ datanode.setPersistedOpState(IN_SERVICE);
+ supervisor.nodeStateUpdated(
+ HddsProtos.NodeOperationalState.DECOMMISSIONING);
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ int expected = repConf.scaleOutOfServiceLimit(2);
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(expected,
+ pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolResizeDuringDecommission(ContainerLayoutVersion
layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 2);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ try {
+ datanode.setPersistedOpState(IN_SERVICE);
+ supervisor.nodeStateUpdated(DECOMMISSIONING);
+ supervisor.setPerVolumePoolSize(2);
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ int expected = repConf.scaleOutOfServiceLimit(2);
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(expected,
+ pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void nonPushReplicationUsesGlobalPoolWhenPerVolumeEnabled(
+ ContainerLayoutVersion layout, @TempDir File perVolumeTempDir) throws
Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+ AtomicInteger globalExecutions = new AtomicInteger();
+
+ ExecutorService trackingGlobal = new AbstractExecutorService() {
+ @Override
+ public void shutdown() {
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return emptyList();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return false;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return false;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return true;
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ globalExecutions.incrementAndGet();
+ command.run();
+ }
+ };
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(trackingGlobal)
+ .clock(clock)
+ .build();
+
+ try {
+ supervisor.addTask(createReconciliationTask(1L));
+ assertEquals(1, globalExecutions.get());
+ assertEquals(1, supervisor.getReplicationSuccessCount());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePushIsolation(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ HddsVolume vol1 = (HddsVolume) volumeSet.getVolumesList().get(0);
+ HddsVolume vol2 = (HddsVolume) volumeSet.getVolumesList().get(1);
+
+ addContainerOnVolume(1L, vol1, conf);
+ addContainerOnVolume(2L, vol2, conf);
+
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ CountDownLatch vol1Started = new CountDownLatch(1);
+ CountDownLatch vol1Release = new CountDownLatch(1);
+ ContainerReplicator volumeAwareReplicator = task -> {
+ Container<?> container = set.getContainer(task.getContainerId());
+ HddsVolume volume = container.getContainerData().getVolume();
+ if (volume == vol1) {
+ vol1Started.countDown();
+ try {
+ assertTrue(vol1Release.await(10, TimeUnit.SECONDS));
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(ie);
+ }
+ }
+ task.setStatus(DONE);
+ };
+ replicatorRef.set(volumeAwareReplicator);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ try {
+ supervisor.addTask(createPushTask(1L));
+ assertTrue(vol1Started.await(10, TimeUnit.SECONDS));
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationSuccessCount() >= 1, 100, 10000);
+
+ assertEquals(1, supervisor.getReplicationSuccessCount());
+ vol1Release.countDown();
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationSuccessCount() == 2, 100, 10000);
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void volumeFailureCleansUpQueuedTasks(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ HddsVolume vol1 = (HddsVolume) volumeSet.getVolumesList().get(0);
+ addContainerOnVolume(1L, vol1, conf);
+ addContainerOnVolume(2L, vol1, conf);
+
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ CountDownLatch task1Started = new CountDownLatch(1);
+ CountDownLatch task1Block = new CountDownLatch(1);
+ replicatorRef.set(task -> {
+ if (task.getContainerId() == 1L) {
+ task1Started.countDown();
+ try {
+ assertTrue(task1Block.await(30, TimeUnit.SECONDS));
Review Comment:
For example, the worker can record the expected interrupt and the JUnit
thread can assert it:
```diff
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java
@@
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@
CountDownLatch task1Started = new CountDownLatch(1);
CountDownLatch task1Block = new CountDownLatch(1);
+ AtomicBoolean task1Interrupted = new AtomicBoolean();
replicatorRef.set(task -> {
if (task.getContainerId() == 1L) {
task1Started.countDown();
try {
- assertTrue(task1Block.await(30, TimeUnit.SECONDS));
+ task1Block.await();
} catch (InterruptedException ie) {
+ task1Interrupted.set(true);
Thread.currentThread().interrupt();
- throw new AssertionError(ie);
+ return;
}
}
task.setStatus(DONE);
@@
GenericTestUtils.waitFor((BooleanSupplier) () ->
supervisor.getTotalInFlightReplications() == 0, 100, 5000);
+ assertTrue(task1Interrupted.get());
task1Block.countDown();
```
The `return` prevents the interrupted task from being marked `DONE`, so the
later success wait still proves that the fallback task executed.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationServer.java:
##########
@@ -183,16 +183,27 @@ public static final class ReplicationConfig {
static final String REPLICATION_OUTOFSERVICE_FACTOR_KEY =
PREFIX + "." + OUTOFSERVICE_FACTOR_KEY;
+ public static final String PER_VOLUME_ENABLED_KEY =
+ PREFIX + ".per.volume.enabled";
+ public static final String PER_VOLUME_STREAMS_LIMIT_KEY =
+ PREFIX + ".per.volume.streams.limit";
+ public static final int PER_VOLUME_STREAMS_LIMIT_DEFAULT = 2;
+
/**
- * The maximum number of replication commands a single datanode can execute
- * simultaneously.
+ * Maximum concurrent tasks on the global replication handler thread pool.
*/
@Config(key = "hdds.datanode.replication.streams.limit",
type = ConfigType.INT,
defaultValue = "10",
tags = {DATANODE},
- description = "The maximum number of replication commands a single " +
- "datanode can execute simultaneously"
+ description = "Maximum concurrent replication tasks on the global "
Review Comment:
One possible documentation-only diff:
```diff
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationServer.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationServer.java
@@
- /**
- * Maximum concurrent tasks on the global replication handler thread
pool.
- */
+ /**
+ * Base size of the global replication handler executor and inbound
+ * replication server executor.
+ */
@@
- description = "Maximum concurrent replication tasks on the global "
- + "replication handler thread pool (before "
- + "outofservice.limit.factor scaling). When "
+ description = "Sets both the base size of the global replication "
+ + "handler executor and the inbound replication server
executor. "
+ + "The global executor is subject to outofservice.limit.factor "
+ + "scaling. When "
+ "hdds.datanode.replication.per.volume.enabled is false
(default), "
- + "all replication tasks use this pool. When per.volume.enabled
is "
- + "true, this pool handles non-push replication and push tasks "
- + "that fall back from a missing per-volume pool; push
concurrency "
- + "per disk is governed by per.volume.streams.limit instead."
+ + "all source-side replication tasks use the global executor. "
+ + "When per.volume.enabled is true, per-volume executors handle
"
+ + "normal source-side push tasks, while this limit still
applies "
+ + "to non-push and fallback source tasks and target-side
inbound "
+ + "push requests."
diff --git a/hadoop-hdds/docs/content/feature/Decommission.md
b/hadoop-hdds/docs/content/feature/Decommission.md
@@
* **`hdds.datanode.replication.streams.limit`**
- * **Purpose**: Sets the base size of the **global** replication
handler thread pool on a DataNode.
+ * **Purpose**: Sets the base size of both the global replication
handler executor and the inbound replication server executor.
* **Default**: `10`.
- * **Details**: For decommissioning nodes, this limit is scaled by
`hdds.datanode.replication.outofservice.limit.factor`. When
`hdds.datanode.replication.per.volume.enabled` is false (default), all
replication tasks use this pool. When per-volume push pools are enabled, this
pool serves non-push replication and push tasks that fall back to the global
executor; per-disk push concurrency is set separately (below).
+ * **Details**: On decommissioning nodes, the global executor is
scaled by `hdds.datanode.replication.outofservice.limit.factor`. Per-volume
pools replace normal source-side push scheduling, but target-side inbound push
requests remain limited by the inbound replication server executor configured
by this property.
diff --git a/hadoop-hdds/docs/content/feature/Reconfigurability.md
b/hadoop-hdds/docs/content/feature/Reconfigurability.md
@@
-| `hdds.datanode.replication.streams.limit` | `10` | Global replication
handler pool size (all tasks when per-volume push pools are disabled) |
+| `hdds.datanode.replication.streams.limit` | `10` | Base size of the
global replication handler executor and inbound replication server executor |
```
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java:
##########
@@ -318,7 +320,9 @@ public String getNamespace() {
.register(OZONE_BLOCK_DELETING_SERVICE_TIMEOUT,
this::reconfigBlockDeletingServiceTimeout)
.register(REPLICATION_STREAMS_LIMIT_KEY,
- this::reconfigReplicationStreamsLimit);
+ this::reconfigReplicationStreamsLimit)
+ .register(PER_VOLUME_STREAMS_LIMIT_KEY,
Review Comment:
A minimal fix would be:
```diff
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/reconfig/TestDatanodeReconfiguration.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/reconfig/TestDatanodeReconfiguration.java
@@
import static
org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration.HDDS_DATANODE_BLOCK_DELETE_THREAD_MAX;
+import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.PER_VOLUME_STREAMS_LIMIT_KEY;
import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.REPLICATION_STREAMS_LIMIT_KEY;
@@
.add(OZONE_BLOCK_DELETING_SERVICE_TIMEOUT)
+ .add(PER_VOLUME_STREAMS_LIMIT_KEY)
.add(REPLICATION_STREAMS_LIMIT_KEY)
```
--
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]
