Copilot commented on code in PR #10693: URL: https://github.com/apache/ozone/pull/10693#discussion_r3574445564
########## 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: volumeLabel is extracted using lastIndexOf('/'), which is not portable (eg Windows paths use '\\'). This can lead to thread names containing full paths or unexpected characters. Prefer deriving the label via java.io.File#getName() (or Paths.get(...).getFileName()) so it works for all path formats. ########## hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java: ########## @@ -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; Review Comment: The PR description states that after a volume failure “new push tasks for containers on that volume are rejected”, but resolveVolumeExecutor falls back to the global executor when the container volume is failed (or when the per-volume pool is missing). This means new push replications can still be queued/executed globally for failed volumes, which contradicts the stated behavior. ########## hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java: ########## @@ -1052,4 +1070,439 @@ 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(); + assertDoesNotThrow(() -> vol1Release.await(10, TimeUnit.SECONDS)); + } Review Comment: This latch wait only asserts that no exception was thrown, but it does not assert that the latch was actually released (await can return false on timeout). That can make the cross-volume isolation test pass even if vol1Release is never counted down within the timeout. ########## hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java: ########## @@ -1052,4 +1070,439 @@ 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(); + assertDoesNotThrow(() -> vol1Release.await(10, TimeUnit.SECONDS)); + } + 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); + replicatorRef.set(task -> { + if (task.getContainerId() == 1L) { + task1Started.countDown(); + assertDoesNotThrow(() -> task1Block.await(30, TimeUnit.SECONDS)); + } Review Comment: This latch wait only asserts that no exception was thrown, but it does not assert that the latch was actually released (await can return false on timeout). If the thread is not unblocked within the timeout, the test may continue and pass without validating the intended blocking/interrupt behavior. -- 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]
