jojochuang commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3574434088
##########
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:
falling back to the global queue is a workaround to deal with HDDS-15327.
Without fixing HDDS-15327, the rejected tasks will stay rejected until SCM's
internal timer times out and start another batch. Will address this later.
--
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]
