Copilot commented on code in PR #10703:
URL: https://github.com/apache/ozone/pull/10703#discussion_r3561880610
##########
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() {
+ return cmd.getTargetDatanode() != null;
+ }
Review Comment:
`ReplicationTask.isPushReplication()` currently always returns true because
`ReplicateContainerCommand` requires a non-null target (see
`Objects.requireNonNull(target)` in its constructor). Keeping a runtime
null-check here is misleading and suggests there are non-push
`ReplicationTask`s when there are not.
##########
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:
When a per-volume executor is unavailable (eg volume failed) `addToQueue`
rejects the task but does not increment the replication failure counters. This
means failures can be silently dropped (no `failureCounter` / `requestCounter`
update), and tests waiting on `getReplicationFailureCount()` will time out.
Additionally, `taskExecutor.execute(...)` can throw
`RejectedExecutionException` if the pool is concurrently shut down (eg during
`shutdownFailedVolumePools`), leaving the task stuck in `inFlight` with
counters never decremented.
--
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]
