atiaomar1978-hub commented on code in PR #26028:
URL: https://github.com/apache/camel/pull/26028#discussion_r3910873283
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -151,60 +162,105 @@ private BackgroundTask createTask() {
.build();
}
- private void onLeadershipTaken() throws Exception {
+ private void onLeadershipTaken() {
lock.lock();
try {
if (!isRunAllowed()) {
return;
}
- if (delegatedConsumer != null) {
+ leadershipTaken = true;
+
+ if (delegatedConsumer != null || isStartPending()) {
return;
}
- final BackgroundTask leaderTask = createTask();
- leaderTask.schedule(getEndpoint().getCamelContext(), () -> {
- if (!isRunAllowed()) {
- return false;
- }
- LOG.info("Leadership taken. Attempt #{} to start consumer:
{}", leaderTask.iteration(), delegatedEndpoint);
-
- Exception cause = null;
- try {
- if (delegatedConsumer == null) {
- delegatedConsumer =
delegatedEndpoint.createConsumer(processor);
- if (delegatedConsumer instanceof StartupListener) {
-
getEndpoint().getCamelContext().addStartupListener((StartupListener)
delegatedConsumer);
- }
- if (delegatedConsumer instanceof ResumeAware
resumeAwareConsumer && resumeStrategy != null) {
- LOG.debug("Setting up the resume adapter for the
resume strategy in consumer");
- ResumeAdapter resumeAdapter
- =
AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer,
- resumeStrategy);
- resumeStrategy.setAdapter(resumeAdapter);
-
- LOG.debug("Setting up the resume strategy for
consumer");
-
resumeAwareConsumer.setResumeStrategy(resumeStrategy);
- }
- }
- ServiceHelper.startService(delegatedEndpoint,
delegatedConsumer);
+ // a task from a previous leadership term may still be scheduled,
drop it
+ cancelLeaderTask(false);
+
Review Comment:
**Possible resource leak (non-blocking):** When leadership is lost after
`createConsumer()` succeeded but before start completes, this path returns
without shutting down `attempt.get()`. Consider cleaning up the attempt
consumer when aborting due to `!leadershipTaken`.
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -151,60 +162,105 @@ private BackgroundTask createTask() {
.build();
}
- private void onLeadershipTaken() throws Exception {
+ private void onLeadershipTaken() {
lock.lock();
try {
if (!isRunAllowed()) {
return;
}
- if (delegatedConsumer != null) {
+ leadershipTaken = true;
+
+ if (delegatedConsumer != null || isStartPending()) {
return;
}
- final BackgroundTask leaderTask = createTask();
- leaderTask.schedule(getEndpoint().getCamelContext(), () -> {
- if (!isRunAllowed()) {
- return false;
- }
- LOG.info("Leadership taken. Attempt #{} to start consumer:
{}", leaderTask.iteration(), delegatedEndpoint);
-
- Exception cause = null;
- try {
- if (delegatedConsumer == null) {
- delegatedConsumer =
delegatedEndpoint.createConsumer(processor);
- if (delegatedConsumer instanceof StartupListener) {
-
getEndpoint().getCamelContext().addStartupListener((StartupListener)
delegatedConsumer);
- }
- if (delegatedConsumer instanceof ResumeAware
resumeAwareConsumer && resumeStrategy != null) {
- LOG.debug("Setting up the resume adapter for the
resume strategy in consumer");
- ResumeAdapter resumeAdapter
- =
AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer,
- resumeStrategy);
- resumeStrategy.setAdapter(resumeAdapter);
-
- LOG.debug("Setting up the resume strategy for
consumer");
-
resumeAwareConsumer.setResumeStrategy(resumeStrategy);
- }
- }
- ServiceHelper.startService(delegatedEndpoint,
delegatedConsumer);
+ // a task from a previous leadership term may still be scheduled,
drop it
+ cancelLeaderTask(false);
+
+ final BackgroundTask task = createTask();
+ // the consumer is created once and re-used by the start attempts
of this task
+ final AtomicReference<Consumer> attempt = new AtomicReference<>();
+ leaderTaskFuture = task.schedule(getEndpoint().getCamelContext(),
() -> startDelegatedConsumer(task, attempt));
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private boolean startDelegatedConsumer(BackgroundTask task,
AtomicReference<Consumer> attempt) {
+ try {
+ // interruptibly, so cancelling the task while this consumer is
being stopped does not
+ // keep the leader pool thread waiting for a lock the stopping
thread holds
Review Comment:
**StartupListener registration:** `addStartupListener` runs once when the
attempt consumer is first created and is reused across retry attempts. If
leadership is lost mid-retry, the listener can remain registered even though
`delegatedConsumer` is never published. Worth a brief code comment so this is
not mistaken for a leak in a later refactor.
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -213,6 +269,15 @@ private void onLeadershipTaken() throws Exception {
private void onLeadershipLost() {
lock.lock();
try {
+ leadershipTaken = false;
+ // a start scheduled by the leadership taken event may not have
run yet, cancel it so it
+ // cannot start the consumer on a node that is no longer the leader
+ cancelLeaderTask(false);
+
+ if (delegatedConsumer == null) {
+ return;
+ }
+
LOG.debug("Leadership lost. Stopping consumer: {}",
delegatedEndpoint);
try {
ServiceHelper.stopAndShutdownServices(delegatedConsumer,
delegatedEndpoint);
Review Comment:
**Good fix for event ordering.** Dispatching `onLeadershipLost` even when
`delegatedConsumer` is still null (pending start) is the core of CAMEL-24583.
Serialising under the consumer lock prevents concurrent leadership events from
being applied out of order. The cluster-view lock-ordering note in the PR
description is appreciated — agree it is pre-existing and out of scope here.
##########
core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java:
##########
@@ -154,8 +161,25 @@ private void runTaskWrapper(CamelContext camelContext,
BooleanSupplier supplier)
*/
public Future<?> schedule(CamelContext camelContext, BooleanSupplier
supplier) {
running.set(true);
- return service.scheduleWithFixedDelay(() ->
runTaskWrapper(camelContext, supplier), budget.initialDelay(),
- budget.interval(), TimeUnit.MILLISECONDS);
+ Future<?> future = service.scheduleWithFixedDelay(() ->
runTaskWrapper(camelContext, supplier),
+ budget.initialDelay(), budget.interval(),
TimeUnit.MILLISECONDS);
+ scheduledFuture.set(future);
+ if (latch.getCount() == 0) {
+ // the task already finished before the future was published, so
it could not unschedule itself
+ unschedule();
+ }
+ return future;
Review Comment:
**CAMEL-24584 looks correct.** Auto-`unschedule()` on completed/exhausted
paths stops the no-op rescheduling that made `isStartPending()` unreliable on
cancelled futures. The post-`schedule()` latch check handles the
synchronous-finish race. Upgrade-guide entry documents the
`Future.isCancelled()` behaviour change.
--
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]
