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: [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]
