Copilot commented on code in PR #10703:
URL: https://github.com/apache/ozone/pull/10703#discussion_r3562294007
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -266,17 +302,68 @@ public void initCounters(AbstractReplicationTask task) {
}
private void addToQueue(AbstractReplicationTask task) {
- if (inFlight.add(task)) {
- if (task.getPriority() != ReplicationCommandPriority.LOW) {
- // Low priority tasks are not included in the replication queue sizes
- // returned to SCM in the heartbeat, so we only update the count for
- // priorities other than low.
- taskCounter.computeIfAbsent(task.getClass(),
- k -> new AtomicInteger()).incrementAndGet();
- }
- queuedCounter.get(task.getMetricName()).incrementAndGet();
- executor.execute(new TaskRunner(task));
+ if (!inFlight.add(task)) {
+ return;
+ }
+ if (task.getPriority() != ReplicationCommandPriority.LOW) {
+ taskCounter.computeIfAbsent(task.getClass(),
+ k -> new AtomicInteger()).incrementAndGet();
}
+ queuedCounter.get(task.getMetricName()).incrementAndGet();
+ ExecutorService taskExecutor = selectExecutor(task);
+ if (taskExecutor == null) {
+ LOG.warn("Rejected {} in ReplicationSupervisor: no replication handler "
+ + "thread pool available", task);
+ queuedCounter.get(task.getMetricName()).decrementAndGet();
+ inFlight.remove(task);
+ decrementTaskCounter(task);
+ return;
+ }
Review Comment:
When per-volume pools are enabled and selectExecutor() returns null (eg
volume failed / pool removed), the task is rejected before TaskRunner runs, so
replication failure metrics never increment. This can cause callers/tests
waiting on getReplicationFailureCount() to time out and makes operational
metrics under-report failures. Consider marking the task FAILED and
incrementing request/failure counters when rejecting due to missing executor.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java:
##########
@@ -204,9 +204,16 @@ public DatanodeStateMachine(HddsDatanodeService
hddsDatanodeService,
.stateContext(context)
.datanodeConfig(dnConf)
.replicationConfig(replicationConfig)
+ .containerSet(container.getContainerSet())
+ .volumeSet(container.getVolumeSet())
.clock(clock)
.build();
+ container.getVolumeSet().setFailedVolumeListener(() -> {
+ container.handleVolumeFailures();
+ supervisor.shutdownFailedVolumePools(container.getVolumeSet());
+ });
Review Comment:
The failed-volume listener now wraps container.handleVolumeFailures() and
supervisor.shutdownFailedVolumePools(). If handleVolumeFailures() throws,
shutdownFailedVolumePools() will be skipped, leaving the per-volume pool for a
failed volume alive. Use try/finally so the pool shutdown still runs on
failures.
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java:
##########
@@ -1052,4 +1069,348 @@ 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 {
+ supervisor.stop();
+ }
Review Comment:
LogCapturer instances add a Log4j appender and must be stopped; otherwise
captured appenders can leak into subsequent tests and duplicate log output.
Call stopCapturing() for both capturers in the finally block.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationServer.java:
##########
@@ -224,6 +230,29 @@ public static final class ReplicationConfig {
)
private double outOfServiceFactor = OUTOFSERVICE_FACTOR_DEFAULT;
+ @Config(key = PER_VOLUME_ENABLED_KEY,
+ type = ConfigType.BOOLEAN,
+ defaultValue = "false",
+ tags = {DATANODE},
+ description = "When true, push-based container replication uses a " +
+ "separate replication handler thread pool per data volume so " +
+ "that slow replication on one disk does not block replication " +
+ "on other disks. Pull replication and other replication tasks " +
+ "continue to use the global replication handler thread pool."
+ )
+ private boolean perVolumeEnabled = false;
+
+ @Config(key = PER_VOLUME_STREAMS_LIMIT_KEY,
+ type = ConfigType.INT,
+ defaultValue = "1",
+ reconfigurable = true,
+ tags = {DATANODE},
+ description = "The maximum number of concurrent push replication " +
+ "commands per data volume when per-volume replication thread " +
+ "pools are enabled."
+ )
+ private int perVolumeStreamsLimit = PER_VOLUME_STREAMS_LIMIT_DEFAULT;
Review Comment:
The PR description says "Test only" / focuses on adding integration tests,
but this change set also introduces new production config keys and runtime
behavior for per-volume replication thread pools (eg PER_VOLUME_* in
ReplicationConfig, ReplicationSupervisor routing, datanode reconfig handler).
Please update the PR description/title (or split PRs) so reviewers/CI
expectations match the actual scope.
--
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]
