From d4d84b8362282b73575541634dfaac566149b100 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Thu, 30 Jul 2026 01:39:37 +0100 Subject: [PATCH 1/7] remove infinite timeout on http lookup --- .../cluster/bootstrap/SelfAwareJoinDecider.scala | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index 4ae8d4dea..b0e589998 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -47,13 +47,19 @@ import scala.concurrent.duration._ * to HTTP binding, during [[pekko.management.scaladsl.PekkoManagement.start()]], hence we * accept blocking on this initialization. If no value is received, the future will fail with * a `TimeoutException` and ClusterBootstrap will log an explanatory error to the user. + * + * The result is cached after the first successful resolution to avoid repeated blocking. */ - private[bootstrap] def selfContactPoint: (String, Int) = - Await.result( + @volatile private var cachedSelfContactPoint: Option[(String, Int)] = None + + private[bootstrap] def selfContactPoint: (String, Int) = cachedSelfContactPoint.getOrElse { + val result = Await.result( ClusterBootstrap(system).selfContactPoint .map(uri => (uri.authority.host.toString, uri.authority.port))(system.dispatcher), - Duration.Inf // the future has a timeout - ) + 15.seconds) + cachedSelfContactPoint = Some(result) + result + } /** * Determines whether it has the need and ability to join self and create a new cluster. @@ -66,7 +72,7 @@ import scala.concurrent.duration._ log.warning( BootstrapLogMarker.inProgress(info.contactPoints.map(contactPointString), info.allSeedNodes), "Self contact point [{}] not found in targets {}", - contactPointString(selfContactPoint), + contactPointString(self), info.contactPoints.mkString(", ")) } false From 986b8a446104542bf553beefeca3662f1d5967b0 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 4 Aug 2026 22:32:27 +0100 Subject: [PATCH 2/7] Update SelfAwareJoinDecider.scala --- .../cluster/bootstrap/SelfAwareJoinDecider.scala | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index b0e589998..272d0d68b 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -48,18 +48,14 @@ import scala.concurrent.duration._ * accept blocking on this initialization. If no value is received, the future will fail with * a `TimeoutException` and ClusterBootstrap will log an explanatory error to the user. * - * The result is cached after the first successful resolution to avoid repeated blocking. + * The result is cached after the first successful resolution. If the resolution fails, + * the next access will retry. */ - @volatile private var cachedSelfContactPoint: Option[(String, Int)] = None - - private[bootstrap] def selfContactPoint: (String, Int) = cachedSelfContactPoint.getOrElse { - val result = Await.result( + private[bootstrap] lazy val selfContactPoint: (String, Int) = + Await.result( ClusterBootstrap(system).selfContactPoint .map(uri => (uri.authority.host.toString, uri.authority.port))(system.dispatcher), 15.seconds) - cachedSelfContactPoint = Some(result) - result - } /** * Determines whether it has the need and ability to join self and create a new cluster. From a303e89a9771198268b4e6f07e6f26c067e8b837 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 4 Aug 2026 22:34:22 +0100 Subject: [PATCH 3/7] Update SelfAwareJoinDecider.scala --- .../management/cluster/bootstrap/SelfAwareJoinDecider.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index 272d0d68b..f2049ae34 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -55,7 +55,7 @@ import scala.concurrent.duration._ Await.result( ClusterBootstrap(system).selfContactPoint .map(uri => (uri.authority.host.toString, uri.authority.port))(system.dispatcher), - 15.seconds) + 30.seconds) /** * Determines whether it has the need and ability to join self and create a new cluster. From 7324afb67e4377e3ab1c46c019380ac16c99cc20 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 25 Aug 2026 19:13:28 +0100 Subject: [PATCH 4/7] Link the self contact point timeouts, cache resolution, and test both Motivation: Follow-up on the review of this branch. The 30 second wait added here and the 10 second timer in `ClusterBootstrap.ensureSelfContactPoint` are coupled - the wait is only correct while it outlasts the timer that completes the promise - but nothing expressed that. Raising the timer would silently make the decider time out first, replacing the deliberate "'Bootstrap.selfContactPoint' was NOT set" error with a bare TimeoutException from elsewhere. The `lazy val` re-runs its initialiser after a throw, so in the case this change exists for - `start()` never ran, so nothing ever completes the promise - every `canJoinSelf` call blocks a dispatcher thread for the full timeout again. The previous `Duration.Inf` parked one thread once; this parks one per probe, for as long as the contact point stays unset. There was no test. `SelfAwareJoinDeciderSpec` only covers the path where the contact point has already been set, and a 30 second hardcoded timeout cannot be exercised in a test anyway. Modification: Move the timer's duration to `ClusterBootstrap.SelfContactPointTimeout` and derive the decider's wait from it, behind a `protected def` a test can override. Replace the `lazy val` with an `AtomicReference` that caches a resolved value. A failure is deliberately not cached, so a contact point set later is still picked up, but the blocking wait is paid at most once: reaching the timeout means the promise has nothing to complete it, so later callers check the promise without blocking and fail fast until it does complete. Resolve against the promise directly rather than mapping it first, so that the already-completed case needs no dispatcher hop and `value` is meaningful. Result: The two timeouts cannot drift apart. An unset contact point costs one blocking wait rather than one per probe, and is still picked up if it arrives late. Tests: - sbt "management-cluster-bootstrap/test" - 54 succeeded, 0 failed (48 before) - New SelfContactPointResolutionSpec covers resolution, caching, the timeout, the block-once behaviour, late setting after a timeout, and the ordering between the two timeouts - Directional: with the block-once guard removed so that every call waits, "block for the timeout only once, then fail fast" FAILS with "505796658 nanoseconds was not less than 250 milliseconds" - sbt "management-cluster-bootstrap/mimaReportBinaryIssues" - success - sbt "management-cluster-bootstrap/scalafmtCheck" "management-cluster-bootstrap/Test/scalafmtCheck", "+headerCheckAll" - clean References: Refs #908 --- .../cluster/bootstrap/ClusterBootstrap.scala | 24 ++- .../bootstrap/SelfAwareJoinDecider.scala | 49 +++++- .../SelfContactPointResolutionSpec.scala | 146 ++++++++++++++++++ 3 files changed, 205 insertions(+), 14 deletions(-) create mode 100644 management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala index a48340712..9e76240c9 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala @@ -126,14 +126,15 @@ final class ClusterBootstrap(implicit system: ExtendedActorSystem) extends Exten * * We give the required selfContactPoint some time to be set asynchronously, or else log an error. */ - @InternalApi private[bootstrap] def ensureSelfContactPoint(): Unit = system.scheduler.scheduleOnce(10.seconds) { - if (!selfContactPoint.isCompleted) { - _selfContactPointUri.failure(new TimeoutException("Awaiting Bootstrap.selfContactPoint timed out.")) - log.error( - "'Bootstrap.selfContactPoint' was NOT set, but is required for the bootstrap to work " + - "if binding bootstrap routes manually and not via pekko-management.") + @InternalApi private[bootstrap] def ensureSelfContactPoint(): Unit = + system.scheduler.scheduleOnce(ClusterBootstrap.SelfContactPointTimeout) { + if (!selfContactPoint.isCompleted) { + _selfContactPointUri.failure(new TimeoutException("Awaiting Bootstrap.selfContactPoint timed out.")) + log.error( + "'Bootstrap.selfContactPoint' was NOT set, but is required for the bootstrap to work " + + "if binding bootstrap routes manually and not via pekko-management.") + } } - } /** * INTERNAL API @@ -153,6 +154,15 @@ final class ClusterBootstrap(implicit system: ExtendedActorSystem) extends Exten object ClusterBootstrap extends ExtensionId[ClusterBootstrap] with ExtensionIdProvider { + /** + * INTERNAL API + * + * How long `selfContactPoint` is given to be set before the promise behind it is failed. Anything + * waiting on that promise has to outlast this, so it is shared rather than restated - see + * `SelfAwareJoinDecider.selfContactPointTimeout`. + */ + @InternalApi private[bootstrap] val SelfContactPointTimeout: FiniteDuration = 10.seconds + override def lookup: ClusterBootstrap.type = ClusterBootstrap override def get(system: ActorSystem): ClusterBootstrap = super.get(system) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index f2049ae34..376a12369 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -13,11 +13,15 @@ package org.apache.pekko.management.cluster.bootstrap +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.{ AtomicBoolean, AtomicReference } + import org.apache.pekko import pekko.actor.ActorSystem import pekko.annotation.InternalApi import pekko.discovery.ServiceDiscovery.ResolvedTarget import pekko.event.{ LogSource, Logging } +import pekko.http.scaladsl.model.Uri import scala.concurrent.Await import scala.concurrent.duration._ @@ -42,20 +46,51 @@ import scala.concurrent.duration._ protected def contactPointString(contactPoint: ResolvedTarget): String = s"${contactPoint.host}:${contactPoint.port.getOrElse("0")}" + /** + * How long to block waiting for the self contact point to be set. `ClusterBootstrap` fails the + * promise itself after [[ClusterBootstrap.SelfContactPointTimeout]], so this only has to outlast + * that; it is derived from it rather than restated so that the two cannot drift apart. Overridable + * for tests, which cannot afford to wait this long. + */ + protected def selfContactPointTimeout: FiniteDuration = ClusterBootstrap.SelfContactPointTimeout * 3 + + private val cachedSelfContactPoint = new AtomicReference[(String, Int)]() + private val hasBlockedForSelfContactPoint = new AtomicBoolean(false) + /** * The value `ClusterBootstrap(system).selfContactPoints` is set prior * to HTTP binding, during [[pekko.management.scaladsl.PekkoManagement.start()]], hence we * accept blocking on this initialization. If no value is received, the future will fail with * a `TimeoutException` and ClusterBootstrap will log an explanatory error to the user. * - * The result is cached after the first successful resolution. If the resolution fails, - * the next access will retry. + * A resolved value is cached. A failure is not, so that a contact point set later is still picked + * up, but the blocking wait is only ever paid once: once `ClusterBootstrap.start()` has run, the + * promise is always completed within [[ClusterBootstrap.SelfContactPointTimeout]], so reaching + * the timeout at all means it never ran and no amount of further waiting will help. */ - private[bootstrap] lazy val selfContactPoint: (String, Int) = - Await.result( - ClusterBootstrap(system).selfContactPoint - .map(uri => (uri.authority.host.toString, uri.authority.port))(system.dispatcher), - 30.seconds) + private[bootstrap] def selfContactPoint: (String, Int) = { + val cached = cachedSelfContactPoint.get() + if (cached ne null) cached + else { + val pending = ClusterBootstrap(system).selfContactPoint + val uri = pending.value match { + case Some(completed) => completed.get + case None if hasBlockedForSelfContactPoint.compareAndSet(false, true) => + Await.result(pending, selfContactPointTimeout) + case None => + throw new TimeoutException( + "'Bootstrap.selfContactPoint' is still not set after waiting " + + s"[$selfContactPointTimeout] for it once. It is required for the bootstrap to work " + + "if binding bootstrap routes manually and not via pekko-management.") + } + val resolved = toContactPoint(uri) + cachedSelfContactPoint.compareAndSet(null, resolved) + resolved + } + } + + private def toContactPoint(uri: Uri): (String, Int) = + (uri.authority.host.toString, uri.authority.port) /** * Determines whether it has the need and ability to join self and create a new cluster. diff --git a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala new file mode 100644 index 000000000..2f54fbbdc --- /dev/null +++ b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala @@ -0,0 +1,146 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.management.cluster.bootstrap + +import java.util.concurrent.TimeoutException + +import com.typesafe.config.ConfigFactory +import org.apache.pekko +import pekko.actor.ActorSystem +import pekko.event.NoLogging +import pekko.testkit.TestKit +import org.scalatest.BeforeAndAfterAll +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import scala.concurrent.duration._ + +object SelfContactPointResolutionSpec { + + /** Waits a fraction of the real timeout, so that the blocking path can be exercised in a test. */ + class ShortTimeoutJoinDecider(system: ActorSystem, settings: ClusterBootstrapSettings) + extends LowestAddressJoinDecider(system, settings) { + override protected def selfContactPointTimeout: FiniteDuration = 500.millis + } + + /** Reads back the production timeout, which is otherwise protected. */ + class TimeoutReadingJoinDecider(system: ActorSystem, settings: ClusterBootstrapSettings) + extends LowestAddressJoinDecider(system, settings) { + def configuredTimeout: FiniteDuration = selfContactPointTimeout + } + +} + +class SelfContactPointResolutionSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll { + + import SelfContactPointResolutionSpec._ + + private val config = ConfigFactory.parseString(""" + pekko { + loglevel = INFO + remote.artery.canonical.port = 0 + management.http { + hostname = "10.0.0.2" + base-path = "test" + } + } + """).withFallback(ConfigFactory.load()) + + // a system per test, because setSelfContactPoint completes a promise that cannot be reset + private var systems: List[ActorSystem] = Nil + + private def newDecider(name: String): (ActorSystem, ShortTimeoutJoinDecider) = { + val system = ActorSystem(name, config) + systems = system :: systems + val settings = ClusterBootstrapSettings(system.settings.config, NoLogging) + (system, new ShortTimeoutJoinDecider(system, settings)) + } + + private def elapsed(body: => Unit): FiniteDuration = { + val started = System.nanoTime() + body + (System.nanoTime() - started).nanos + } + + "SelfAwareJoinDecider.selfContactPoint" should { + + "resolve the contact point once it has been set" in { + val (system, decider) = newDecider("self-contact-point-resolves") + ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") + + decider.selfContactPoint should ===(("10.0.0.2", 8558)) + } + + "cache the resolved contact point rather than resolving it again" in { + val (system, decider) = newDecider("self-contact-point-caches") + ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") + + val first = decider.selfContactPoint + val second = decider.selfContactPoint + + second should ===(first) + // the cache holds the same instance, not just an equal one + second.asInstanceOf[AnyRef] should be theSameInstanceAs first.asInstanceOf[AnyRef] + } + + "time out instead of blocking forever when the contact point is never set" in { + val (_, decider) = newDecider("self-contact-point-times-out") + + a[TimeoutException] should be thrownBy decider.selfContactPoint + } + + "block for the timeout only once, then fail fast" in { + val (_, decider) = newDecider("self-contact-point-fails-fast") + + val firstCall = elapsed(a[TimeoutException] should be thrownBy decider.selfContactPoint) + val secondCall = elapsed(a[TimeoutException] should be thrownBy decider.selfContactPoint) + + firstCall should be >= 500.millis + // without this the bootstrap coordinator would park a dispatcher thread for the full + // timeout on every probe, for as long as the contact point stays unset + secondCall should be < 250.millis + } + + "still pick up a contact point that is set after a timeout" in { + val (system, decider) = newDecider("self-contact-point-set-late") + + a[TimeoutException] should be thrownBy decider.selfContactPoint + + ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") + + decider.selfContactPoint should ===(("10.0.0.2", 8558)) + } + + } + + "ClusterBootstrap.SelfContactPointTimeout" should { + + "be outlasted by the decider's own wait, so that the promise fails first" in { + val (system, _) = newDecider("self-contact-point-timeout-ordering") + val settings = ClusterBootstrapSettings(system.settings.config, NoLogging) + val decider = new TimeoutReadingJoinDecider(system, settings) + + decider.configuredTimeout should be > ClusterBootstrap.SelfContactPointTimeout + } + + } + + override def afterAll(): Unit = + systems.foreach(TestKit.shutdownActorSystem(_, 5.seconds)) + +} From 870d191995ae35dcce1bac78c40361bf2b06b659 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 25 Aug 2026 19:34:12 +0100 Subject: [PATCH 5/7] Apply scalafmt to SelfContactPointResolutionSpec --- .../cluster/bootstrap/SelfContactPointResolutionSpec.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala index 2f54fbbdc..b4f3cea9e 100644 --- a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala +++ b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala @@ -95,7 +95,7 @@ class SelfContactPointResolutionSpec extends AnyWordSpec with Matchers with Befo second should ===(first) // the cache holds the same instance, not just an equal one - second.asInstanceOf[AnyRef] should be theSameInstanceAs first.asInstanceOf[AnyRef] + assert(second eq first) } "time out instead of blocking forever when the contact point is never set" in { From 39c20cc8518ed35261be2d40f05407d4ac1b6e2d Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 25 Aug 2026 19:40:42 +0100 Subject: [PATCH 6/7] Do not scaladoc-link a private member Motivation: ClusterBootstrap.SelfContactPointTimeout is private[bootstrap], so scaladoc cannot resolve a [[...]] link to it and unidoc fails the Docs compile job with 'Could not find any member to link'. Modification: Refer to it as code rather than as a link, in both places. Result: Scaladoc generates again. Tests: - sbt "unidoc; docs/paradox" - success, which is the exact command the Docs compile job runs References: Refs #908 --- .../management/cluster/bootstrap/SelfAwareJoinDecider.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index 376a12369..328e5ba34 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -48,7 +48,7 @@ import scala.concurrent.duration._ /** * How long to block waiting for the self contact point to be set. `ClusterBootstrap` fails the - * promise itself after [[ClusterBootstrap.SelfContactPointTimeout]], so this only has to outlast + * promise itself after `ClusterBootstrap.SelfContactPointTimeout`, so this only has to outlast * that; it is derived from it rather than restated so that the two cannot drift apart. Overridable * for tests, which cannot afford to wait this long. */ @@ -65,7 +65,7 @@ import scala.concurrent.duration._ * * A resolved value is cached. A failure is not, so that a contact point set later is still picked * up, but the blocking wait is only ever paid once: once `ClusterBootstrap.start()` has run, the - * promise is always completed within [[ClusterBootstrap.SelfContactPointTimeout]], so reaching + * promise is always completed within `ClusterBootstrap.SelfContactPointTimeout`, so reaching * the timeout at all means it never ran and no amount of further waiting will help. */ private[bootstrap] def selfContactPoint: (String, Int) = { From 7ebf810b933cf59a387c98e8eb31532e7f5b467b Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Sat, 29 Aug 2026 21:55:28 +0100 Subject: [PATCH 7/7] Do not let a synchronous decide() failure wedge the coordinator Motivation: `SelfAwareJoinDecider.selfContactPoint` resolves eagerly, so `LowestAddressJoinDecider.decide` can throw before it ever constructs a Future. `BootstrapCoordinator.decide()` only guards the returned Future with `recover`, which never sees a synchronous throw. The exception escapes with `decisionInProgress` already set to true, and that flag is only reset on receipt of a JoinDecision, so every later DecideTick short-circuits on "Previous decision still in progress" and the coordinator stops deciding for the rest of its life. This is reachable once the decider's wait no longer outlasts the promise's own timeout, which is exactly what the fail-fast path introduced here can do. Modification: - Wrap `joinDecider.decide(info)` so a NonFatal synchronous throw becomes a failed Future, letting the existing `recover` log it and fall back to KeepProbing, which resets the flag. - Explain why the two fields in SelfAwareJoinDecider are atomic, given that in-tree callers are already serialised by the coordinator, and what each one holds. - Assert the caching test's actual property rather than tuple reference equality, which pinned an implementation detail. Result: An unset self contact point can no longer stop cluster bootstrap permanently; the coordinator keeps probing and recovers if the contact point is set later. Tests: - New BootstrapCoordinatorSpec case "keep making decisions instead of wedging". Directional: with the try/catch removed it fails with "1 was not greater than or equal to 2" - a single decide() call, then silence. - sbt "management-cluster-bootstrap/test" - 55 succeeded, 0 failed. - sbt "management-cluster-bootstrap/mimaReportBinaryIssues" - success. - scalafmt --mode diff-ref=origin/main and sbt +headerCheckAll - clean. References: Refs #908 --- .../bootstrap/SelfAwareJoinDecider.scala | 8 +++ .../internal/BootstrapCoordinator.scala | 15 ++++-- .../SelfContactPointResolutionSpec.scala | 5 +- .../internal/BootstrapCoordinatorSpec.scala | 52 ++++++++++++++++++- 4 files changed, 72 insertions(+), 8 deletions(-) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index 328e5ba34..7d79ef969 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -54,7 +54,15 @@ import scala.concurrent.duration._ */ protected def selfContactPointTimeout: FiniteDuration = ClusterBootstrap.SelfContactPointTimeout * 3 + // `decide` is called from the BootstrapCoordinator actor and guarded by its `decisionInProgress` + // flag, so in-tree callers are already serialised. `JoinDecider` is a user-pluggable interface + // though, and a custom one is free to resolve the contact point inside the Future it returns, so + // these stay atomic rather than plain vars. + // + // `cachedSelfContactPoint` holds the resolved value, `null` until there is one. Failures are not + // cached, so a contact point set later is still picked up. private val cachedSelfContactPoint = new AtomicReference[(String, Int)]() + // Set by whichever caller takes the one permitted blocking wait; everyone else fails fast. private val hasBlockedForSelfContactPoint = new AtomicBoolean(false) /** diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala index dba790253..8cece9d68 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala @@ -16,7 +16,7 @@ package org.apache.pekko.management.cluster.bootstrap.internal import java.time.LocalDateTime import java.util.concurrent.ThreadLocalRandom import scala.collection.immutable -import scala.concurrent.ExecutionContext +import scala.concurrent.{ ExecutionContext, Future } import org.apache.pekko import pekko.actor.Actor import pekko.actor.ActorRef @@ -34,6 +34,7 @@ import pekko.pattern.pipe import scala.concurrent.duration._ import scala.util.Try +import scala.util.control.NonFatal import pekko.event.Logging import pekko.management.cluster.bootstrap.{ BootstrapLogMarker, @@ -392,8 +393,16 @@ private[pekko] class BootstrapCoordinator( decisionInProgress = true - joinDecider - .decide(info) + // `decide` may throw synchronously before it ever builds a Future (it resolves the self + // contact point eagerly), and `recover` would not see that. Letting it escape would leave + // `decisionInProgress` stuck true and stop this coordinator deciding ever again. + val decision = + try joinDecider.decide(info) + catch { + case NonFatal(e) => Future.failed(e) + } + + decision .recover { case e => log.error(e, "Join decision failed: {}", e) diff --git a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala index b4f3cea9e..f14ada4c1 100644 --- a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala +++ b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala @@ -86,16 +86,15 @@ class SelfContactPointResolutionSpec extends AnyWordSpec with Matchers with Befo decider.selfContactPoint should ===(("10.0.0.2", 8558)) } - "cache the resolved contact point rather than resolving it again" in { + "keep returning the resolved contact point on later calls" in { val (system, decider) = newDecider("self-contact-point-caches") ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") val first = decider.selfContactPoint val second = decider.selfContactPoint + first should ===(("10.0.0.2", 8558)) second should ===(first) - // the cache holds the same instance, not just an equal one - assert(second eq first) } "time out instead of blocking forever when the contact point is never set" in { diff --git a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala index 9995694cb..7591ed363 100644 --- a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala +++ b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala @@ -13,7 +13,8 @@ package org.apache.pekko.management.cluster.bootstrap.internal -import java.util.concurrent.atomic.AtomicReference +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.{ AtomicInteger, AtomicReference } import org.apache.pekko import pekko.actor.{ ActorRef, ActorSystem, Props } @@ -21,7 +22,12 @@ import pekko.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget } import pekko.discovery.{ Lookup, MockDiscovery } import pekko.http.scaladsl.model.Uri import pekko.management.cluster.bootstrap.internal.BootstrapCoordinator.Protocol.InitiateBootstrapping -import pekko.management.cluster.bootstrap.{ ClusterBootstrapSettings, LowestAddressJoinDecider } +import pekko.management.cluster.bootstrap.{ + ClusterBootstrapSettings, + JoinDecision, + LowestAddressJoinDecider, + SeedNodesInformation +} import com.typesafe.config.ConfigFactory import org.scalatest.BeforeAndAfterAll import org.scalatest.concurrent.Eventually @@ -177,6 +183,48 @@ class BootstrapCoordinatorSpec extends AnyWordSpec with Matchers with BeforeAndA } } + "The bootstrap coordinator, when a JoinDecider throws synchronously" should { + + // SelfAwareJoinDecider.selfContactPoint resolves eagerly and can throw before `decide` ever + // builds a Future, so the failure never reaches `recover`. If that escapes, `decisionInProgress` + // is left true and the coordinator stops deciding for the rest of its life. + "keep making decisions instead of wedging" in { + val decideServiceName = "bootstrap-coordinator-throwing-decider" + val decideSettings = ClusterBootstrapSettings( + ConfigFactory.parseString(s""" + |pekko.management.cluster.bootstrap { + | contact-point-discovery.service-name = $decideServiceName + | contact-point-discovery.required-contact-point-nr = 1 + | contact-point-discovery.contact-with-all-contact-points = false + |} + """.stripMargin).withFallback(system.settings.config), + system.log) + + MockDiscovery.set( + Lookup(decideServiceName, portName = None, protocol = Some("tcp")), + () => Future.successful(Resolved(decideServiceName, List(ResolvedTarget("host1", Some(7626), None))))) + + val decideCalls = new AtomicInteger(0) + val throwingDecider = new LowestAddressJoinDecider(system, decideSettings) { + override def decide(info: SeedNodesInformation): Future[JoinDecision] = { + decideCalls.incrementAndGet() + throw new TimeoutException("'Bootstrap.selfContactPoint' was never set") + } + } + + val coordinator = system.actorOf(Props(new BootstrapCoordinator(discovery, throwingDecider, decideSettings) { + override def ensureProbing(selfContactPointScheme: String, contactPoint: ResolvedTarget): Option[ActorRef] = + None + })) + coordinator ! InitiateBootstrapping(selfUri) + + // more than one call proves decisionInProgress was reset after the throw + eventually { + decideCalls.get should be >= 2 + } + } + } + override def afterAll(): Unit = { Await.result(system.terminate(), 10.seconds) super.afterAll()