Copilot commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3574530821
##########
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) {
+ 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);
+ }
Review Comment:
Per-volume pools are initialized based on `perVolumeEnabled` + `volumeSet`,
but `containerSet` can still be null. In that case the pools are created but
will never be selected (resolveVolumeExecutor always falls back to the global
executor), which wastes threads and makes the feature silently ineffective in
misconfigured builds/tests. Consider only initializing per-volume pools when
both `volumeSet` and `containerSet` are present (and log a warning otherwise).
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -304,9 +394,49 @@ public void stop() {
executor.shutdownNow();
}
Review Comment:
If `stop()` has to fall back to `shutdownNow()`, the drained-but-never-run
tasks are currently left in `inFlight` / `queuedCounter` / `taskCounter`
(unlike the per-volume pools which explicitly clean up drained `TaskRunner`s).
Capturing the `shutdownNow()` return list and passing it through
`cancelDrainedTaskRunners` keeps metrics/state consistent during shutdown and
makes tests less prone to leaking counters.
##########
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 com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.File;
+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) {
+ File storageDir = volume.getStorageDir();
+ String volumeRoot = storageDir.getPath();
+ volumeRoots.add(volumeRoot);
+ pools.put(volumeRoot,
+ createPool(poolSize, threadNamePrefix, storageDir.getName()));
+ }
Review Comment:
Thread names for per-volume pools use only `storageDir.getName()` as the
volume label. If multiple volumes share the same leaf directory name (eg
`/disk1/hdds` and `/disk2/hdds`), threads from different pools will have
identical names, making thread dumps and log correlation ambiguous. Consider
including a stable unique suffix derived from the full volume path (eg a hash)
in the thread name label.
--
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]
