Copilot commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3574395787
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -191,8 +213,20 @@ public ReplicationSupervisor build() {
};
}
+ if (replicationConfig.isPerVolumeEnabled() && volumeSet != null) {
Review Comment:
Per-volume pools are initialized when per-volume is enabled and a volumeSet
is provided, but routing relies on containerSet. If containerSet is null, this
will still spin up per-volume executors that are never used (selectExecutor
falls back to the global executor). Consider requiring both containerSet and
volumeSet before enabling per-volume pools.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -266,17 +304,70 @@ 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: replication handler "
+ + "thread pool unavailable", task, e);
+ 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;
Review Comment:
The PR description says that after a volume failure the per-volume pool is
shut down and new push tasks for containers on that volume are rejected, but
this code falls back to the global replication handler thread pool when the
volume is failed or its per-volume pool is missing. That means failed-volume
tasks may continue to run in the global pool, which contradicts the documented
behavior and could reintroduce cross-volume interference. Please either update
the description/tests, or change this path to reject tasks for failed volumes
(eg by throwing RejectedExecutionException so addToQueue rolls back counters).
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/VolumeReplicationThreadPools.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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 com.google.common.util.concurrent.ThreadFactoryBuilder;
+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 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) {
+ String volumeLabel = volumeRoot.substring(
+ Math.max(0, volumeRoot.lastIndexOf('/') + 1));
Review Comment:
Thread name volume label extraction uses a hard-coded '/' separator, which
will not work correctly on non-POSIX path separators. Using File/Path APIs
avoids OS-specific assumptions.
--
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]
