smengcl commented on code in PR #10693: URL: https://github.com/apache/ozone/pull/10693#discussion_r3561998169
########## 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(); Review Comment: `shutdownNow()` discards the queued `Runnable`s it returns. Those queued `TaskRunner`s never execute, so the `finally` in `ReplicationSupervisor.TaskRunner.run()` that calls `inFlight.remove(task)` and decrements the counters never runs for them. Their `(containerId, target)` entries stay in `inFlight` permanently, and since `addToQueue` early returns when `inFlight.add(task)` is false, every later push replication command for those containers is dropped until the datanode restarts. This fires exactly in the failed volume path this feature adds. Consider draining the returned list and running the same cleanup, for example have `shutdownVolume` return `pool.shutdownNow()` so `shutdownFailedVolumePools` can call back into the supervisor to remove each drained task from `inFlight` and fix the counters. Otherwise a single volume failure can strand re-replication for its containers. ########## hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java: ########## @@ -304,9 +384,36 @@ public void stop() { executor.shutdownNow(); } } catch (InterruptedException ie) { - // Ignore, we don't really care about the failure. Thread.currentThread().interrupt(); } + if (volumePools != null) { + volumePools.shutdownAll(); + } + } + + public ReplicationConfig getReplicationConfig() { + return replicationConfig; + } + + public void setPerVolumePoolSize(int newSize) { + if (volumePools != null) { + volumePools.setPoolSize(newSize); Review Comment: This writes the raw `newSize` straight to the pools and skips the out-of-service scaling that `resize()` applies (see the per-volume block in `resize`). `setReplicationMaxStreams` sets the config then calls `resize(state.get())`, so the global pool gets `newSize * outOfServiceFactor` while decommissioning or in maintenance. Reconfiguring `per.volume.streams.limit` on a node that is already DECOMMISSIONING leaves the per-volume pools at the unscaled value, which is inconsistent with the global pool during the exact scenario this feature targets. Suggest mirroring `setReplicationMaxStreams`: set `replicationConfig.setPerVolumeStreamsLimit(newSize)` first, then call `resize(state.get())` to reuse the existing scaling path instead of calling `setPoolSize` directly. ########## hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java: ########## @@ -105,6 +105,14 @@ DatanodeDetails getTarget() { return cmd.getTargetDatanode(); } + /** + * @return true if this is a push replication task (source datanode sends + * container to target). + */ + public boolean isPushReplication() { Review Comment: `ReplicateContainerCommand` calls `Objects.requireNonNull(targetDatanode)` in every constructor and in `getFromProtobuf`, so `getTargetDatanode()` is never null and this always returns true. The guard `if (!replicationTask.isPushReplication()) { return executor; }` in `ReplicationSupervisor.selectExecutor` is therefore unreachable today. If this is meant as a guard for a future pull style task, a short comment saying so would help; otherwise it can be dropped to avoid implying a branch that never runs. -- 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]
