henrik242 commented on code in PR #26028:
URL: https://github.com/apache/camel/pull/26028#discussion_r3916251553
##########
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
+ lock.lockInterruptibly();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return true; // no more attempts
+ }
+ try {
+ if (!isRunAllowed()) {
+ return false;
+ }
+
+ if (!leadershipTaken) {
+ // leadership was lost while this start was pending. Starting
now would run the consumer on a
+ // node that is not the leader, and no further leadership
event is coming to stop it again
+ LOG.debug("Leadership lost while the start was pending. Not
starting consumer: {}", delegatedEndpoint);
+ return true; // no more attempts
+ }
- } catch (Exception e) {
- cause = e;
+ if (delegatedConsumer != null) {
+ return true; // no more attempts
+ }
+
+ LOG.info("Leadership taken. Attempt #{} to start consumer: {}",
task.iteration(), delegatedEndpoint);
+
+ Exception cause = null;
+ try {
+ Consumer consumer = attempt.get();
+ if (consumer == null) {
+ consumer = delegatedEndpoint.createConsumer(processor);
+ attempt.set(consumer);
+ if (consumer instanceof StartupListener startupListener) {
+
getEndpoint().getCamelContext().addStartupListener(startupListener);
+ }
+ if (consumer 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, consumer);
+ // publish the consumer only once it is started, so a failed
attempt cannot leave an unstarted
+ // consumer behind that makes every later leadership term
believe there is nothing left to do
+ delegatedConsumer = consumer;
+
+ } catch (Exception e) {
+ cause = e;
+ }
- if (cause != null) {
- String message = "Leadership taken. Attempt #" +
leaderTask.iteration()
- + " failed to start consumer due to: " +
cause.getMessage();
- getExceptionHandler().handleException(message, cause);
- // make the task runner aware of the exception (will retry)
- throw new TaskRunFailureException(message, cause);
+ if (cause != null) {
+ String message = "Leadership taken. Attempt #" +
task.iteration()
+ + " failed to start consumer due to: " +
cause.getMessage();
+ getExceptionHandler().handleException(message, cause);
+ if (task.iteration() >=
masterEndpoint.getComponent().getBackOffMaxAttempts()) {
Review Comment:
confirmed, withUnlimitedDuration() added, log guarded by maxAttempts > 0,
new test that fails without the one-line fix.
--
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]
