You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "jojochuang (via GitHub)" <gi...@apache.org> on 2026/07/07 05:09:48 UTC
Re: [PR] HDDS-15412. [DataNode] Disk volume-specific container replication thread pool [ozone]
jojochuang commented on code in PR #10595:
URL: https://github.com/apache/ozone/pull/10595#discussion_r3532428662
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java:
##########
@@ -221,6 +224,23 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
.datanodeConfig(dnConf)
.replicationConfig(replicationConfig)
.clock(clock)
+ .volumeChooser(task -> {
+ if (task instanceof ReplicationTask) {
+ ReplicationTask repTask = (ReplicationTask) task;
+ if (repTask.getTarget() == null) {
Review Comment:
pull-based replication
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java:
##########
@@ -221,6 +224,23 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
.datanodeConfig(dnConf)
.replicationConfig(replicationConfig)
.clock(clock)
+ .volumeChooser(task -> {
+ if (task instanceof ReplicationTask) {
+ ReplicationTask repTask = (ReplicationTask) task;
+ if (repTask.getTarget() == null) {
+ try {
+ return importer.chooseNextVolume(importer.getDefaultReplicationSpace());
+ } catch (IOException e) {
+ LOG.error("Failed to choose volume for replication task " + task, e);
+ return null;
+ }
+ } else {
Review Comment:
pull-based replication
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java:
##########
@@ -72,11 +72,13 @@ public void replicate(ReplicationTask task) {
LOG.info("Starting replication of container {} from {} using {}",
containerID, sourceDatanodes, compression);
- HddsVolume targetVolume = null;
-
+ HddsVolume targetVolume = task.getVolume();
try {
- targetVolume = containerImporter.chooseNextVolume(
- containerImporter.getDefaultReplicationSpace());
+ if (targetVolume == null) {
Review Comment:
DownloadAndImportReplication is used by pull replication
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -283,7 +307,70 @@ private void addToQueue(AbstractReplicationTask task) {
k -> new AtomicInteger()).incrementAndGet();
}
queuedCounter.get(task.getMetricName()).incrementAndGet();
- executor.execute(new TaskRunner(task));
+ getExecutorForTask(task).execute(new TaskRunner(task));
+ }
+ }
+
+ private ExecutorService getExecutorForTask(AbstractReplicationTask task) {
+ cleanupFailedVolumeExecutors();
+ HddsVolume volume = task.getVolume();
+ if (volume != null) {
+ return getOrCreateVolumeExecutor(volume);
+ }
+ return executor;
+ }
+
+ private synchronized ThreadPoolExecutor getOrCreateVolumeExecutor(HddsVolume volume) {
+ return volumeExecutors.computeIfAbsent(volume, v -> {
+ String threadNamePrefix = context != null ? context.getThreadNamePrefix() : "";
+ ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat(threadNamePrefix + "ContainerReplicationThread-" + v.getStorageID() + "-%d")
Review Comment:
better to use volume root to make troubleshooting easier
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java:
##########
@@ -205,20 +211,36 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
importer,
new SimpleContainerDownloader(conf, certClient));
ContainerReplicator pushReplicator = new PushReplicator(conf,
- new OnDemandContainerReplicationSource(container.getController()),
+ new OnDemandContainerReplicationSource(container.getController(),
+ replicationConfig),
new GrpcContainerUploader(conf, certClient, container.getController())
);
pullReplicatorWithMetrics = new MeasuredReplicator(pullReplicator, "pull");
pushReplicatorWithMetrics = new MeasuredReplicator(pushReplicator, "push");
- ReplicationConfig replicationConfig =
- conf.getObject(ReplicationConfig.class);
supervisor = ReplicationSupervisor.newBuilder()
.stateContext(context)
.datanodeConfig(dnConf)
.replicationConfig(replicationConfig)
.clock(clock)
+ .volumeChooser(task -> {
+ if (task instanceof ReplicationTask) {
+ ReplicationTask repTask = (ReplicationTask) task;
+ if (repTask.getTarget() == null) {
+ try {
+ return importer.chooseNextVolume(importer.getDefaultReplicationSpace());
Review Comment:
This is a reasonable ask. Though push-replication is the default and pull mode is planned to be deprecated/removed in the future.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org