Copilot commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3574371108
##########
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;
}
+ return volumeExecutor;
Review Comment:
When per-volume replication is enabled and the container’s volume is failed
(or the per-volume pool has been removed), resolveVolumeExecutor currently
falls back to the global executor. This contradicts the PR description (“new
push tasks for containers on that volume are rejected”) and can reintroduce
cross-volume interference by running push replication for a failed/slow disk in
the global pool. Consider rejecting these tasks via RejectedExecutionException
so addToQueue can roll back counters and drop the task.
##########
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);
+ }
Review Comment:
ReplicationSupervisor logs RejectedExecutionException with a full stack
trace. In this codebase, RejectedExecutionException is often an expected
condition (eg queue full / pool shutdown) and is typically logged without the
exception to avoid noisy logs. Logging just the message (and keeping the
rollback) should be sufficient.
##########
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));
+ }
+ task.setStatus(DONE);
+ });
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ String volumeRoot = vol1.getStorageDir().getPath();
+ try {
+ supervisor.addTask(createPushTask(1L));
+ assertTrue(task1Started.await(10, TimeUnit.SECONDS));
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getTotalInFlightReplications() == 2, 100, 5000);
+
+ volumeSet.failVolume(volumeRoot);
+ supervisor.shutdownFailedVolumePools(volumeSet);
+
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getTotalInFlightReplications() == 0, 100, 5000);
+ task1Block.countDown();
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationSuccessCount() >= 1, 100, 5000);
+ } finally {
+ task1Block.countDown();
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void volumeFailureShutsDownPool(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);
+
+ 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();
+ replicatorRef.set(doneReplicator);
+
+ String volumeRoot = vol1.getStorageDir().getPath();
+ try {
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ assertTrue(pools.hasPool(volumeRoot));
+
+ volumeSet.failVolume(volumeRoot);
+ supervisor.shutdownFailedVolumePools(volumeSet);
+ assertFalse(pools.hasPool(volumeRoot));
+
+ supervisor.addTask(createPushTask(1L));
+ assertEquals(1, supervisor.getReplicationSuccessCount());
Review Comment:
After failing the volume and shutting down its per-volume pool, this test
expects a subsequent push task on that volume to succeed. If the intended
behavior is to reject new push tasks for containers on a failed volume, the
assertion should verify that the task does not execute (eg success count
remains 0 and in-flight remains 0).
##########
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));
+ }
+ task.setStatus(DONE);
+ });
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ String volumeRoot = vol1.getStorageDir().getPath();
+ try {
+ supervisor.addTask(createPushTask(1L));
+ assertTrue(task1Started.await(10, TimeUnit.SECONDS));
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getTotalInFlightReplications() == 2, 100, 5000);
+
+ volumeSet.failVolume(volumeRoot);
+ supervisor.shutdownFailedVolumePools(volumeSet);
+
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getTotalInFlightReplications() == 0, 100, 5000);
+ task1Block.countDown();
+
+ supervisor.addTask(createPushTask(2L));
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationSuccessCount() >= 1, 100, 5000);
Review Comment:
This test currently expects a push task submitted after volume failure/pool
shutdown to succeed. That matches the current fallback-to-global behavior, but
conflicts with the stated intent that new push tasks for a failed volume are
rejected. If the supervisor rejects tasks for failed volumes, assert rejection
indirectly by checking in-flight/success counters stay unchanged after
addTask().
--
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]
