This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 0c4c66e9f4b HDDS-15412. Add per-volume push replication thread pools
on DataNode.
0c4c66e9f4b is described below
commit 0c4c66e9f4b4b7bf84d6135235b2d0ae0d74dd92
Author: Wei-Chiu Chuang <[email protected]>
AuthorDate: Fri Jul 31 19:12:30 2026 -0700
HDDS-15412. Add per-volume push replication thread pools on DataNode.
Co-authored-by: Cursor <[email protected]>
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../apache/hadoop/ozone/HddsDatanodeService.java | 28 +-
.../common/statemachine/DatanodeStateMachine.java | 7 +
.../container/replication/ReplicationServer.java | 73 +++-
.../replication/ReplicationSupervisor.java | 183 +++++++-
.../replication/ReplicationSupervisorMetrics.java | 5 +-
.../replication/VolumeReplicationThreadPools.java | 166 ++++++++
.../replication/TestReplicationConfig.java | 30 ++
.../replication/TestReplicationSupervisor.java | 467 +++++++++++++++++++++
hadoop-hdds/docs/content/feature/Decommission.md | 9 +-
.../docs/content/feature/Reconfigurability.md | 3 +-
.../reconfig/TestDatanodeReconfiguration.java | 2 +
11 files changed, 947 insertions(+), 26 deletions(-)
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
index c2a30d97d52..356e5887745 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
@@ -30,6 +30,7 @@
import static org.apache.hadoop.ozone.common.Storage.StorageState.INITIALIZED;
import static
org.apache.hadoop.ozone.conf.OzoneServiceConfig.DEFAULT_SHUTDOWN_HOOK_PRIORITY;
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;
import static org.apache.hadoop.security.UserGroupInformation.getCurrentUser;
import static org.apache.hadoop.util.ExitUtil.terminate;
@@ -96,6 +97,7 @@
import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
import
org.apache.hadoop.ozone.container.diskbalancer.DiskBalancerProtocolServer;
+import
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig;
import org.apache.hadoop.ozone.ha.ConfUtils;
import org.apache.hadoop.ozone.util.OzoneNetUtils;
import org.apache.hadoop.ozone.util.ShutdownHookManager;
@@ -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,
+ this::reconfigPerVolumeStreamsLimit);
scmServiceId = HddsUtils.getScmServiceId(conf);
@@ -719,6 +723,28 @@ private String reconfigReplicationStreamsLimit(String
value) {
return value;
}
+ private String reconfigPerVolumeStreamsLimit(String value) {
+ int newSize = Integer.parseInt(value);
+ Preconditions.checkArgument(newSize >= 1,
+ PER_VOLUME_STREAMS_LIMIT_KEY + " must be at least 1 but was %s",
+ value);
+ ReplicationConfig replicationConfig =
+ getDatanodeStateMachine().getSupervisor().getReplicationConfig();
+ if (!replicationConfig.isPerVolumeEnabled()) {
+ LOG.warn("Ignoring reconfiguration of {} to {} because per-volume "
+ + "replication is disabled", PER_VOLUME_STREAMS_LIMIT_KEY, value);
+ return value;
+ }
+ try {
+ getDatanodeStateMachine().getSupervisor().setPerVolumePoolSize(newSize);
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to apply per-volume replication thread pool resize to "
+ + "{}: {}", value, e.getMessage(), e);
+ throw e;
+ }
+ return value;
+ }
+
private String reconfigBlockDeletingServiceInterval(String value) {
return value;
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
index 00a9bdd1043..4f6078d0bd2 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
+++
b/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());
+ });
+
replicationSupervisorMetrics =
ReplicationSupervisorMetrics.create(supervisor);
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
index 8375b1100bd..ceb35201b3c 100644
---
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
@@ -183,16 +183,30 @@ 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.
+ * Base size of the global replication handler executor and inbound
+ * replication server executor.
*/
@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 = "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 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."
)
private int replicationMaxStreams = REPLICATION_MAX_STREAMS_DEFAULT;
@@ -224,6 +238,34 @@ 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 = "2",
+ reconfigurable = true,
+ tags = {DATANODE},
+ description = "When hdds.datanode.replication.per.volume.enabled is "
+ + "true, maximum concurrent push replication commands per data "
+ + "volume (each volume has its own handler thread pool; effective "
+ + "push parallelism on the datanode is roughly the number of "
+ + "volumes times this limit, with outofservice.limit.factor "
+ + "applied per pool on decommissioning or maintenance nodes). "
+ + "Push replication is usually disk-bound, so one or two "
+ + "concurrent transfers per volume often saturates the disk."
+ )
+ private int perVolumeStreamsLimit = PER_VOLUME_STREAMS_LIMIT_DEFAULT;
+
public double getOutOfServiceFactor() {
return outOfServiceFactor;
}
@@ -257,6 +299,22 @@ public void setReplicationQueueLimit(int limit) {
this.replicationQueueLimit = limit;
}
+ public boolean isPerVolumeEnabled() {
+ return perVolumeEnabled;
+ }
+
+ public void setPerVolumeEnabled(boolean enabled) {
+ this.perVolumeEnabled = enabled;
+ }
+
+ public int getPerVolumeStreamsLimit() {
+ return perVolumeStreamsLimit;
+ }
+
+ public void setPerVolumeStreamsLimit(int limit) {
+ this.perVolumeStreamsLimit = limit;
+ }
+
@PostConstruct
public void validate() {
if (replicationMaxStreams < 1) {
@@ -279,6 +337,13 @@ public void validate() {
clamped);
outOfServiceFactor = clamped;
}
+
+ if (perVolumeStreamsLimit < 1) {
+ LOG.warn(PER_VOLUME_STREAMS_LIMIT_KEY + " must be greater than zero " +
+ "and was set to {}. Defaulting to {}",
+ perVolumeStreamsLimit, PER_VOLUME_STREAMS_LIMIT_DEFAULT);
+ perVolumeStreamsLimit = PER_VOLUME_STREAMS_LIMIT_DEFAULT;
+ }
}
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
index 8b6daf6aae5..d805e2249b8 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
@@ -28,6 +28,7 @@
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.OptionalLong;
@@ -35,6 +36,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -49,8 +51,13 @@
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicationCommandPriority;
import org.apache.hadoop.metrics2.lib.MetricsRegistry;
import org.apache.hadoop.metrics2.lib.MutableRate;
+import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
+import org.apache.hadoop.ozone.container.common.interfaces.Container;
import
org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
+import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
+import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
+import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
import
org.apache.hadoop.ozone.container.replication.AbstractReplicationTask.Status;
import
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig;
import org.apache.hadoop.util.Time;
@@ -106,6 +113,8 @@ public final class ReplicationSupervisor {
private final IntConsumer executorThreadUpdater;
private final ReplicationConfig replicationConfig;
private final DatanodeConfiguration datanodeConfig;
+ private final ContainerSet containerSet;
+ private final VolumeReplicationThreadPools volumePools;
/**
* Builder for {@link ReplicationSupervisor}.
@@ -114,10 +123,13 @@ public static class Builder {
private StateContext context;
private ReplicationConfig replicationConfig;
private DatanodeConfiguration datanodeConfig;
+ private ContainerSet containerSet;
+ private MutableVolumeSet volumeSet;
private ExecutorService executor;
private Clock clock;
private IntConsumer executorThreadUpdater = threadCount -> {
};
+ private VolumeReplicationThreadPools volumePools;
public Builder clock(Clock newClock) {
clock = newClock;
@@ -149,6 +161,16 @@ public Builder executorThreadUpdater(IntConsumer
newUpdater) {
return this;
}
+ public Builder containerSet(ContainerSet newContainerSet) {
+ containerSet = newContainerSet;
+ return this;
+ }
+
+ public Builder volumeSet(MutableVolumeSet newVolumeSet) {
+ volumeSet = newVolumeSet;
+ return this;
+ }
+
public ReplicationSupervisor build() {
if (replicationConfig == null || datanodeConfig == null) {
ConfigurationSource conf = new OzoneConfiguration();
@@ -191,8 +213,20 @@ public ReplicationSupervisor build() {
};
}
+ if (replicationConfig.isPerVolumeEnabled() && volumeSet != null) {
+ LOG.info("Per-volume container replication thread pools enabled with "
+ + "{} threads per volume",
+ replicationConfig.getPerVolumeStreamsLimit());
+ volumePools = new VolumeReplicationThreadPools();
+ String threadNamePrefix =
+ context != null ? context.getThreadNamePrefix() : "";
+ volumePools.init(volumeSet.getVolumesList(),
+ replicationConfig.getPerVolumeStreamsLimit(), threadNamePrefix);
+ }
+
return new ReplicationSupervisor(context, executor, replicationConfig,
- datanodeConfig, clock, executorThreadUpdater);
+ datanodeConfig, clock, executorThreadUpdater, containerSet,
+ volumePools);
}
}
@@ -204,14 +238,18 @@ public static Map<String, String> getMetricsMap() {
return Collections.unmodifiableMap(METRICS_MAP);
}
+ @SuppressWarnings("checkstyle:ParameterNumber")
private ReplicationSupervisor(StateContext context, ExecutorService executor,
ReplicationConfig replicationConfig, DatanodeConfiguration
datanodeConfig,
- Clock clock, IntConsumer executorThreadUpdater) {
+ Clock clock, IntConsumer executorThreadUpdater, ContainerSet
containerSet,
+ VolumeReplicationThreadPools volumePools) {
this.inFlight = ConcurrentHashMap.newKeySet();
this.context = context;
this.executor = executor;
this.replicationConfig = replicationConfig;
this.datanodeConfig = datanodeConfig;
+ this.containerSet = containerSet;
+ this.volumePools = volumePools;
maxQueueSize = datanodeConfig.getCommandQueueLimit();
this.clock = clock;
this.executorThreadUpdater = executorThreadUpdater;
@@ -266,17 +304,69 @@ 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();
+ try {
+ selectExecutor(task).execute(new TaskRunner(task));
+ } catch (RejectedExecutionException e) {
+ LOG.warn("Rejected {} in ReplicationSupervisor: {}", task,
e.getMessage());
+ rollbackQueuedTask(task);
+ }
+ }
+
+ private void rollbackQueuedTask(AbstractReplicationTask task) {
+ queuedCounter.get(task.getMetricName()).decrementAndGet();
+ inFlight.remove(task);
+ decrementTaskCounter(task);
+ }
+
+ private ExecutorService selectExecutor(AbstractReplicationTask task) {
+ if (!replicationConfig.isPerVolumeEnabled() || volumePools == null) {
+ return executor;
+ }
+ if (!(task instanceof ReplicationTask)) {
+ return executor;
+ }
+ ReplicationTask replicationTask = (ReplicationTask) task;
+ return resolveVolumeExecutor(replicationTask.getContainerId());
+ }
+
+ private ExecutorService resolveVolumeExecutor(long containerId) {
+ if (containerSet == null) {
+ return executor;
+ }
+ Container<?> container = containerSet.getContainer(containerId);
+ if (container == null) {
+ LOG.warn("Container {} not found for push replication; falling back to "
+ + "ReplicationSupervisor global replication handler thread pool",
+ containerId);
+ return executor;
+ }
+ HddsVolume volume = container.getContainerData().getVolume();
+ String volumeRoot = volume == null ? "unknown"
+ : volume.getStorageDir().getPath();
+ if (volume == null || volume.isFailed()) {
+ LOG.warn("No per-volume replication handler thread pool available for "
+ + "container {} on volume {}; falling back to global replication
"
+ + "handler thread pool",
+ containerId, volumeRoot);
+ return executor;
+ }
+ ExecutorService volumeExecutor = volumePools.getExecutor(volumeRoot);
+ if (volumeExecutor == null) {
+ LOG.warn("No per-volume replication handler thread pool available for "
+ + "container {} on volume {}; falling back to global replication
"
+ + "handler thread pool",
+ containerId, volumeRoot);
+ return executor;
}
+ return volumeExecutor;
}
private void decrementTaskCounter(AbstractReplicationTask task) {
@@ -304,9 +394,49 @@ public void stop() {
executor.shutdownNow();
}
} catch (InterruptedException ie) {
- // Ignore, we don't really care about the failure.
Thread.currentThread().interrupt();
}
+ if (volumePools != null) {
+ cancelDrainedTaskRunners(volumePools.shutdownAll());
+ }
+ }
+
+ public ReplicationConfig getReplicationConfig() {
+ return replicationConfig;
+ }
+
+ public void setPerVolumePoolSize(int newSize) {
+ if (volumePools != null) {
+ replicationConfig.setPerVolumeStreamsLimit(newSize);
+ resize(state.get());
+ }
+ }
+
+ public void shutdownFailedVolumePools(MutableVolumeSet volumeSet) {
+ if (volumePools == null || volumeSet == null) {
+ return;
+ }
+ for (StorageVolume volume : volumeSet.getFailedVolumesList()) {
+ cancelDrainedTaskRunners(
+ volumePools.shutdownVolume(volume.getStorageDir().getPath()));
+ }
+ }
+
+ private void cancelDrainedTaskRunners(List<Runnable> drained) {
+ for (Runnable runnable : drained) {
+ if (!(runnable instanceof TaskRunner)) {
+ continue;
+ }
+ AbstractReplicationTask task = ((TaskRunner) runnable).getTask();
+ queuedCounter.get(task.getMetricName()).decrementAndGet();
+ inFlight.remove(task);
+ decrementTaskCounter(task);
+ }
+ }
+
+ @VisibleForTesting
+ VolumeReplicationThreadPools getVolumeReplicationThreadPools() {
+ return volumePools;
}
/**
@@ -371,6 +501,20 @@ private void resize(HddsProtos.NodeOperationalState
nodeState) {
maxQueueSize = newMaxQueueSize;
executorThreadUpdater.accept(threadCount);
+
+ if (volumePools != null) {
+ int perVolumeThreadCount = replicationConfig.getPerVolumeStreamsLimit();
+ if (isMaintenance(nodeState) || isDecommission(nodeState)) {
+ perVolumeThreadCount =
+ replicationConfig.scaleOutOfServiceLimit(perVolumeThreadCount);
+ }
+ LOG.info("Scaling per-volume replication thread pools to {} "
+ + "(base={}, factor={})",
+ perVolumeThreadCount,
+ replicationConfig.getPerVolumeStreamsLimit(),
+ replicationConfig.getOutOfServiceFactor());
+ volumePools.setPoolSize(perVolumeThreadCount);
+ }
}
/**
@@ -383,6 +527,10 @@ public TaskRunner(AbstractReplicationTask task) {
this.task = task;
}
+ AbstractReplicationTask getTask() {
+ return task;
+ }
+
@Override
public void run() {
final long startTime = Time.monotonicNow();
@@ -478,11 +626,14 @@ public long getReplicationRequestCount(String
metricsName) {
}
public long getQueueSize() {
+ long queueSize = 0;
if (executor instanceof ThreadPoolExecutor) {
- return ((ThreadPoolExecutor)executor).getQueue().size();
- } else {
- return 0;
+ queueSize += ((ThreadPoolExecutor) executor).getQueue().size();
+ }
+ if (volumePools != null) {
+ queueSize += volumePools.getTotalQueueSize();
}
+ return queueSize;
}
public long getMaxReplicationStreams() {
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorMetrics.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorMetrics.java
index 64854e1ea2c..151e07a451b 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorMetrics.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorMetrics.java
@@ -83,8 +83,9 @@ public void getMetrics(MetricsCollector collector, boolean
all) {
"Number of replication requests skipped as the container is "
+ "already present"),
supervisor.getReplicationSkippedCount())
- .addGauge(Interns.info("maxReplicationStreams", "Maximum number of "
- + "concurrent replication tasks which can run simultaneously"),
+ .addGauge(Interns.info("maxReplicationStreams", "Maximum pool size of "
+ + "the global replication handler executor (not total capacity "
+ + "when per-volume push replication thread pools are enabled)"),
supervisor.getMaxReplicationStreams());
Map<String, String> metricsMap = ReplicationSupervisor.getMetricsMap();
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/VolumeReplicationThreadPools.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/VolumeReplicationThreadPools.java
new file mode 100644
index 00000000000..f8ddf8fbaf9
--- /dev/null
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/VolumeReplicationThreadPools.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.replication;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.hdds.utils.HddsServerUtil;
+import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Per-volume replication handler thread pools for push-based replication.
+ */
+final class VolumeReplicationThreadPools {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(VolumeReplicationThreadPools.class);
+
+ private final ConcurrentHashMap<String, ThreadPoolExecutor> pools =
+ new ConcurrentHashMap<>();
+ private int currentPoolSize;
+
+ void init(Collection<? extends StorageVolume> volumes, int poolSize,
+ String threadNamePrefix) {
+ currentPoolSize = poolSize;
+ List<String> volumeRoots = new ArrayList<>();
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ volumeRoots.add(volumeRoot);
+ pools.put(volumeRoot, createPool(poolSize, threadNamePrefix,
volumeRoot));
+ }
+ LOG.info("Initialized {} per-volume replication thread pools "
+ + "(threads per volume = {}): {}",
+ volumeRoots.size(), poolSize, volumeRoots);
+ }
+
+ private static ThreadPoolExecutor createPool(int poolSize,
+ String threadNamePrefix, String volumeRoot) {
+ AtomicInteger threadId = new AtomicInteger();
+ ThreadFactory threadFactory = runnable -> {
+ Thread thread = new Thread(runnable, threadNamePrefix
+ + "ContainerReplicationThread-" + volumeRoot + "-"
+ + threadId.getAndIncrement());
+ thread.setDaemon(true);
+ return thread;
+ };
+ return new ThreadPoolExecutor(
+ poolSize,
+ poolSize,
+ 60, TimeUnit.SECONDS,
+ new PriorityBlockingQueue<>(),
+ threadFactory);
+ }
+
+ ExecutorService getExecutor(String volumeRoot) {
+ return pools.get(volumeRoot);
+ }
+
+ List<Runnable> shutdownVolume(String volumeRoot) {
+ ThreadPoolExecutor pool = pools.remove(volumeRoot);
+ if (pool == null) {
+ return Collections.emptyList();
+ }
+ LOG.info("Shutting down per-volume replication thread pool for failed "
+ + "volume {}", volumeRoot);
+ List<Runnable> drained = Collections.emptyList();
+ try {
+ drained = pool.shutdownNow();
+ if (!pool.awaitTermination(3, TimeUnit.SECONDS)) {
+ LOG.warn("Per-volume replication thread pool for volume {} did not "
+ + "terminate within timeout", volumeRoot);
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while shutting down per-volume replication thread "
+ + "pool for volume {}", volumeRoot, e);
+ Thread.currentThread().interrupt();
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to shut down per-volume replication thread pool for "
+ + "volume {}: {}", volumeRoot, e.getMessage(), e);
+ }
+ return drained;
+ }
+
+ List<Runnable> shutdownAll() {
+ List<Runnable> drained = new ArrayList<>();
+ for (String volumeRoot : new ArrayList<>(pools.keySet())) {
+ drained.addAll(shutdownVolume(volumeRoot));
+ }
+ return drained;
+ }
+
+ void setPoolSize(int newSize) {
+ LOG.info("Resizing per-volume replication thread pools from {} to {}",
+ currentPoolSize, newSize);
+ int successCount = 0;
+ int totalCount = pools.size();
+ for (Map.Entry<String, ThreadPoolExecutor> entry : pools.entrySet()) {
+ try {
+ HddsServerUtil.setPoolSize(entry.getValue(), newSize, LOG);
+ successCount++;
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to resize per-volume replication thread pool for "
+ + "volume {}: {}", entry.getKey(), e.getMessage(), e);
+ }
+ }
+ currentPoolSize = newSize;
+ if (successCount < totalCount) {
+ LOG.warn("Resized {}/{} per-volume replication thread pools to {}",
+ successCount, totalCount, newSize);
+ } else if (totalCount > 0) {
+ LOG.info("Resized all {} per-volume replication thread pools to {}",
+ totalCount, newSize);
+ }
+ }
+
+ int getCurrentPoolSize() {
+ return currentPoolSize;
+ }
+
+ @VisibleForTesting
+ int getPoolSize(String volumeRoot) {
+ ThreadPoolExecutor pool = pools.get(volumeRoot);
+ return pool == null ? 0 : pool.getMaximumPoolSize();
+ }
+
+ @VisibleForTesting
+ long getTotalQueueSize() {
+ long total = 0;
+ for (ThreadPoolExecutor pool : pools.values()) {
+ total += pool.getQueue().size();
+ }
+ return total;
+ }
+
+ @VisibleForTesting
+ boolean hasPool(String volumeRoot) {
+ return pools.containsKey(volumeRoot);
+ }
+}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationConfig.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationConfig.java
index f1f182f40a3..a0b0f544542 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationConfig.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationConfig.java
@@ -20,10 +20,14 @@
import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.OUTOFSERVICE_FACTOR_DEFAULT;
import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.OUTOFSERVICE_FACTOR_MAX;
import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.OUTOFSERVICE_FACTOR_MIN;
+import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.PER_VOLUME_STREAMS_LIMIT_DEFAULT;
+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_MAX_STREAMS_DEFAULT;
import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.REPLICATION_OUTOFSERVICE_FACTOR_KEY;
import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.REPLICATION_STREAMS_LIMIT_KEY;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig;
@@ -138,6 +142,32 @@ public void isCreatedWitDefaultValues() {
subject.getReplicationMaxStreams());
assertEquals(OUTOFSERVICE_FACTOR_DEFAULT,
subject.getOutOfServiceFactor(), 0.001);
+ assertFalse(subject.isPerVolumeEnabled());
+ assertEquals(PER_VOLUME_STREAMS_LIMIT_DEFAULT,
+ subject.getPerVolumeStreamsLimit());
+ }
+
+ @Test
+ public void acceptsPerVolumeConfigValues() {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setBoolean(ReplicationConfig.PER_VOLUME_ENABLED_KEY, true);
+ conf.setInt(PER_VOLUME_STREAMS_LIMIT_KEY, 3);
+
+ ReplicationConfig subject = conf.getObject(ReplicationConfig.class);
+
+ assertTrue(subject.isPerVolumeEnabled());
+ assertEquals(3, subject.getPerVolumeStreamsLimit());
+ }
+
+ @Test
+ public void overridesInvalidPerVolumeStreamsLimit() {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setInt(PER_VOLUME_STREAMS_LIMIT_KEY, 0);
+
+ ReplicationConfig subject = conf.getObject(ReplicationConfig.class);
+
+ assertEquals(PER_VOLUME_STREAMS_LIMIT_DEFAULT,
+ subject.getPerVolumeStreamsLimit());
}
}
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
index 026ac6ac76a..21efce17dee 100644
---
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
@@ -28,9 +28,15 @@
import static
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicationCommandPriority.NORMAL;
import static
org.apache.hadoop.ozone.container.common.impl.ContainerImplTestUtils.newContainerSet;
import static
org.apache.hadoop.ozone.container.replication.AbstractReplicationTask.Status.DONE;
+import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.PER_VOLUME_ENABLED_KEY;
+import static
org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig.PER_VOLUME_STREAMS_LIMIT_KEY;
import static
org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand.toTarget;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyLong;
@@ -59,16 +65,22 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
import java.util.function.Function;
+import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory;
+import org.apache.hadoop.hdds.fs.SpaceUsageCheckFactory;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicationCommandPriority;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.security.symmetric.SecretKeySignerClient;
import
org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
@@ -76,9 +88,13 @@
import org.apache.hadoop.ozone.container.checksum.ReconcileContainerTask;
import org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion;
import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
+import org.apache.hadoop.ozone.container.common.interfaces.Container;
import
org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
import
org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
+import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
+import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
+import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
import
org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCommandInfo;
import
org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCoordinator;
import
org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCoordinatorTask;
@@ -91,6 +107,7 @@
import
org.apache.hadoop.ozone.protocol.commands.ReconstructECContainersCommand;
import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand;
import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ozone.test.GenericTestUtils.LogCapturer;
import org.apache.ozone.test.MockClock;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -107,6 +124,8 @@ public class TestReplicationSupervisor {
private File tempDir;
private final ContainerReplicator noopReplicator = task -> { };
+ private final ContainerReplicator doneReplicator =
+ task -> task.setStatus(DONE);
private final ContainerReplicator throwingReplicator = task -> {
throw new RuntimeException("testing replication failure");
};
@@ -1052,4 +1071,452 @@ 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);
+ AtomicBoolean task1Interrupted = new AtomicBoolean();
+ replicatorRef.set(task -> {
+ if (task.getContainerId() == 1L) {
+ task1Started.countDown();
+ try {
+ task1Block.await();
+ } catch (InterruptedException ie) {
+ task1Interrupted.set(true);
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ task.setStatus(DONE);
+ });
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ String volumeRoot = vol1.getStorageDir().getPath();
+ try {
+ supervisor.addTask(createPushTask(1L));
+ assertTrue(task1Started.await(10, TimeUnit.SECONDS));
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getTotalInFlightReplications() == 2, 100, 5000);
+
+ volumeSet.failVolume(volumeRoot);
+ supervisor.shutdownFailedVolumePools(volumeSet);
+
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getTotalInFlightReplications() == 0, 100, 5000);
+ assertTrue(task1Interrupted.get());
+ task1Block.countDown();
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationSuccessCount() >= 1, 100, 5000);
+ } finally {
+ task1Block.countDown();
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void volumeFailureShutsDownPool(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);
+
+ 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();
+ replicatorRef.set(doneReplicator);
+
+ String volumeRoot = vol1.getStorageDir().getPath();
+ try {
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ assertTrue(pools.hasPool(volumeRoot));
+
+ volumeSet.failVolume(volumeRoot);
+ supervisor.shutdownFailedVolumePools(volumeSet);
+ assertFalse(pools.hasPool(volumeRoot));
+
+ supervisor.addTask(createPushTask(1L));
+ assertEquals(1, supervisor.getReplicationSuccessCount());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ private OzoneConfiguration perVolumeConf(File baseDir, int perVolumeStreams)
{
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, baseDir.getAbsolutePath());
+ conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY,
+ baseDir.getAbsolutePath() + "/vol1,"
+ + baseDir.getAbsolutePath() + "/vol2");
+ conf.setBoolean(PER_VOLUME_ENABLED_KEY, true);
+ conf.setInt(PER_VOLUME_STREAMS_LIMIT_KEY, perVolumeStreams);
+ conf.setClass(SpaceUsageCheckFactory.Conf.configKeyForClassName(),
+ MockSpaceUsageCheckFactory.HalfTera.class,
+ SpaceUsageCheckFactory.class);
+ return conf;
+ }
+
+ private MutableVolumeSet newVolumeSet(OzoneConfiguration conf)
+ throws IOException {
+ return new MutableVolumeSet(datanode.getUuidString(), conf, null,
+ StorageVolume.VolumeType.DATA_VOLUME, null);
+ }
+
+ private void addContainerOnVolume(long containerId, HddsVolume volume,
+ OzoneConfiguration conf) {
+ KeyValueContainerData containerData = new
KeyValueContainerData(containerId,
+ layoutVersion, 100L,
+ UUID.randomUUID().toString(), UUID.randomUUID().toString());
+ containerData.setVolume(volume);
+ KeyValueContainer container = new KeyValueContainer(containerData, conf);
+ assertDoesNotThrow(() -> set.addContainer(container));
+ }
+
+ private ReplicationTask createPushTask(long containerId) {
+ ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget(
+ containerId, MockDatanodeDetails.randomDatanodeDetails());
+ cmd.setTerm(CURRENT_TERM);
+ return new ReplicationTask(cmd, replicatorRef.get());
+ }
}
diff --git a/hadoop-hdds/docs/content/feature/Decommission.md
b/hadoop-hdds/docs/content/feature/Decommission.md
index ede26d6c7e8..53461755f58 100644
--- a/hadoop-hdds/docs/content/feature/Decommission.md
+++ b/hadoop-hdds/docs/content/feature/Decommission.md
@@ -93,9 +93,14 @@ Administrators can adjust the following properties in
`ozone-site.xml` to contro
* **Details**: For decommissioning nodes, this limit is scaled by
`hdds.datanode.replication.outofservice.limit.factor`.
* **`hdds.datanode.replication.streams.limit`**
- * **Purpose**: Sets the base number of threads for the replication
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 also scaled by
`hdds.datanode.replication.outofservice.limit.factor`.
+ * **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.
+
+* **`hdds.datanode.replication.per.volume.streams.limit`**
+ * **Purpose**: When `hdds.datanode.replication.per.volume.enabled` is
true, sets the base number of push replication handler threads **per data
volume**.
+ * **Default**: `2` (reconfigurable at runtime).
+ * **Details**: Each volume has its own pool; total push capacity on the
node scales with the number of volumes. On decommissioning or maintenance
nodes, each per-volume pool is scaled by
`hdds.datanode.replication.outofservice.limit.factor`, same as the global pool.
Push replication is typically disk-bound, so one or two concurrent transfers
per volume is often enough to keep a disk busy while isolating slow volumes.
By tuning these properties, administrators can balance the decommissioning
speed against the impact on the cluster's performance.
diff --git a/hadoop-hdds/docs/content/feature/Reconfigurability.md
b/hadoop-hdds/docs/content/feature/Reconfigurability.md
index 8bfe4a7e46b..4592f9d9485 100644
--- a/hadoop-hdds/docs/content/feature/Reconfigurability.md
+++ b/hadoop-hdds/docs/content/feature/Reconfigurability.md
@@ -104,7 +104,8 @@ ozone admin reconfig --service=[OM|SCM|DATANODE]
--address=<ip:port|hostname:por
| `ozone.block.deleting.service.workers` | `10` | Number of block deletion
service workers |
| `ozone.block.deleting.service.interval` | `60s` | Block deletion service run
interval |
| `ozone.block.deleting.service.timeout` | `300s` | Block deletion service
timeout |
-| `hdds.datanode.replication.streams.limit` | `10` | Maximum replication
streams per datanode |
+| `hdds.datanode.replication.streams.limit` | `10` | Base size of the global
replication handler executor and inbound replication server executor |
+| `hdds.datanode.replication.per.volume.streams.limit` | `2` | Maximum push
replication streams per data volume when per-volume replication thread pools
are enabled (separate from streams.limit) |
## Usage Examples
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
index 071661a9e1c..cf29220bac8 100644
---
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
@@ -21,6 +21,7 @@
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_TIMEOUT;
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_WORKERS;
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;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -53,6 +54,7 @@ void reconfigurableProperties() {
.add(OZONE_BLOCK_DELETING_SERVICE_WORKERS)
.add(OZONE_BLOCK_DELETING_SERVICE_INTERVAL)
.add(OZONE_BLOCK_DELETING_SERVICE_TIMEOUT)
+ .add(PER_VOLUME_STREAMS_LIMIT_KEY)
.add(REPLICATION_STREAMS_LIMIT_KEY)
.addAll(new DatanodeConfiguration().reconfigurableProperties())
.addAll(new TracingConfig().reconfigurableProperties())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
