Copilot commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3548006996
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -266,17 +302,68 @@ 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();
+ ExecutorService taskExecutor = selectExecutor(task);
+ if (taskExecutor == null) {
+ LOG.warn("Rejected {} in ReplicationSupervisor: no replication handler "
+ + "thread pool available", task);
+ queuedCounter.get(task.getMetricName()).decrementAndGet();
+ inFlight.remove(task);
+ decrementTaskCounter(task);
+ return;
+ }
+ taskExecutor.execute(new TaskRunner(task));
+ }
Review Comment:
`taskExecutor.execute(...)` can throw (eg `RejectedExecutionException`) if a
per-volume pool is concurrently shut down due to a volume failure. In that case
the task has already been added to `inFlight` and counters incremented, but the
`TaskRunner` will never run to decrement them, leaving the supervisor
permanently thinking work is in-flight and potentially blocking new tasks. Wrap
the submission in a try/catch and roll back `inFlight` / counters on failure.
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java:
##########
@@ -1052,4 +1069,348 @@ 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 {
+ supervisor.stop();
+ }
Review Comment:
`LogCapturer.captureLogs(...)` requires `stopCapturing()`; otherwise the
appender stays attached and can leak captured output / interfere with later
tests. Add `stopCapturing()` calls in this `finally` block.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/VolumeReplicationThreadPools.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.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));
+ ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat(threadNamePrefix + "ContainerReplicationThread-"
+ + volumeLabel + "-%d")
+ .build();
+ return new ThreadPoolExecutor(
+ poolSize,
+ poolSize,
+ 60, TimeUnit.SECONDS,
+ new PriorityBlockingQueue<>(),
+ threadFactory);
+ }
+
+ ExecutorService getExecutor(String volumeRoot) {
+ return pools.get(volumeRoot);
+ }
+
+ void shutdownVolume(String volumeRoot) {
+ ThreadPoolExecutor pool = pools.remove(volumeRoot);
+ if (pool == null) {
+ return;
+ }
+ LOG.info("Shutting down per-volume replication thread pool for failed "
+ + "volume {}", volumeRoot);
+ try {
+ pool.shutdownNow();
+ if (!pool.awaitTermination(3, TimeUnit.SECONDS)) {
+ LOG.warn("Per-volume replication thread pool for volume {} did not "
Review Comment:
`shutdownNow()` drops any queued `TaskRunner`s for this volume. Those
dropped tasks will never execute their `finally` block in `TaskRunner`, so they
will remain in `ReplicationSupervisor.inFlight` and `queuedCounter`,
potentially blocking replication globally until restart. Consider returning the
dropped runnables from `shutdownVolume` and having `ReplicationSupervisor`
explicitly cancel/cleanup them (or otherwise drain + account for queued tasks)
before discarding the pool.
--
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]
