fix: SBR threshold for down-all-when-indirectly-connected (#32875)

Threshold for when all are downed instead of only the detected indirectly connected, 50% enabled by default

Author: patriknw

.gitignore

@@ -106,3 +106,5 @@ native/
 
 # attachments created by scripts/prepare-downloads.sh
 akka-docs/src/main/paradox/attachments
+
+.claude

akka-cluster/src/main/resources/reference.conf

@@ -426,10 +426,18 @@ akka.cluster.split-brain-resolver {
   # In a malfunctioning network there can be situations where nodes are observed as unreachable
   # via some network links, but they are still indirectly connected via other nodes, i.e. it's
   # not a clean network partition.
-  # By default it will keep fully connected nodes and down all the indirectly connected nodes,
-  # but when this flag is enabled it will down all nodes as precaution for the possible
-  # uncertainty that indirectly connected nodes can cause.
-  down-all-when-indirectly-connected = off
+  # By default it will keep fully connected nodes and down all the indirectly connected nodes.
+  #
+  # However, there is a risk with asymmetric partitions. As a safeguard, if a
+  # DownIndirectlyConnected decision would down more than a certain
+  # fraction of the cluster members, SBR will instead down all nodes.
+  #
+  # The value can be 'on', 'off', or a double threshold between 0.0 and 1.0:
+  # - 'on': Always down all when there are indirectly connected
+  # - 'off': Disables the safeguard entirely
+  # - A double value:  if more or equal to this fraction of members
+  #   would be downed by a DownIndirectlyConnected decision, down all instead
+  down-all-when-indirectly-connected = 0.5
 
 }
 #//#split-brain-resolver

akka-cluster/src/main/scala/akka/cluster/sbr/SplitBrainResolver.scala

@@ -435,16 +435,33 @@ import akka.remote.artery.ThisActorSystemQuarantinedEvent
    */
   def actOnDecision(decision: Decision): Set[UniqueAddress] = {
     val nodesToDown =
-      if (settings.DownAllWhenIndirectlyConnected && decision.isIndirectlyConnected) {
-        strategy.nodesToDown(DownAll)
-      } else {
-        try {
-          strategy.nodesToDown(decision)
-        } catch {
-          case e: IllegalStateException =>
-            log.warning(e.getMessage)
+      try {
+        val nodes = strategy.nodesToDown(decision)
+        // Safeguard for indirectly connected decisions:
+        // If the decision would down more than a threshold fraction of members,
+        // convert to DownAll to prevent split brain from stale gossip in asymmetric partitions.
+        if (decision.isIndirectlyConnected) {
+          val threshold = settings.DownAllWhenIndirectlyConnectedThreshold
+          val memberCount = strategy.members.size
+          if (memberCount > 0 && threshold < 1.0 && nodes.size.toDouble / memberCount >= threshold) {
+            log.warning(
+              ClusterLogMarker.sbrDowning(decision),
+              "SBR indirectly connected decision would down [{}] of [{}] members which is more than " +
+              "the configured down-all-when-indirectly-connected threshold of [{}], downing all instead",
+              nodes.size,
+              memberCount,
+              settings.DownAllWhenIndirectlyConnectedThreshold)
             strategy.nodesToDown(DownAll)
+          } else {
+            nodes
+          }
+        } else {
+          nodes
         }
+      } catch {
+        case e: IllegalStateException =>
+          log.warning(e.getMessage)
+          strategy.nodesToDown(DownAll)
       }
 
     observeDecision(decision, nodesToDown, unreachableDataCenters)

akka-cluster/src/main/scala/akka/cluster/sbr/SplitBrainResolverSettings.scala

@@ -55,10 +55,10 @@ import akka.util.Helpers.Requiring
   val DownAllWhenUnstable: FiniteDuration = {
     val key = "down-all-when-unstable"
     Helpers.toRootLowerCase(cc.getString("down-all-when-unstable")) match {
-      case "on" =>
+      case "on" | "true" =>
         // based on stable-after
         4.seconds.max(DowningStableAfter * 3 / 4)
-      case "off" =>
+      case "off" | "false" =>
         // disabled
         Duration.Zero
       case _ =>
@@ -67,8 +67,20 @@ import akka.util.Helpers.Requiring
     }
   }
 
-  val DownAllWhenIndirectlyConnected: Boolean =
-    cc.getBoolean("down-all-when-indirectly-connected")
+  val DownAllWhenIndirectlyConnectedThreshold: Double = {
+    val key = "down-all-when-indirectly-connected"
+    Helpers.toRootLowerCase(cc.getString(key)) match {
+      case "on" | "true" =>
+        0.0
+      case "off" | "false" =>
+        1.0
+      case _ =>
+        val value = cc.getDouble(key)
+        if (value < 0.0 || value > 1.0)
+          throw new ConfigurationException(s"$key must be 'on', 'off', or a double between 0.0 and 1.0, was [$value]")
+        value
+    }
+  }
 
   // the individual sub-configs below should only be called when the strategy has been selected
 

akka-cluster/src/multi-jvm/scala/akka/cluster/sbr/DownAllIndirectlyConnected5NodeSpec.scala

@@ -14,7 +14,7 @@ import akka.cluster.MultiNodeClusterSpec
 import akka.remote.testkit.Direction
 import akka.remote.testkit.MultiNodeConfig
 
-object DownAllIndirectlyConnected3NodeSpec extends MultiNodeConfig {
+object DownAllWhenIndirectlyConnected3NodeSpec extends MultiNodeConfig {
   val node1 = role("node1")
   val node2 = role("node2")
   val node3 = role("node3")
@@ -40,12 +40,12 @@ object DownAllIndirectlyConnected3NodeSpec extends MultiNodeConfig {
   testTransport(on = true)
 }
 
-class DownAllIndirectlyConnected3NodeSpecMultiJvmNode1 extends DownAllIndirectlyConnected3NodeSpec
-class DownAllIndirectlyConnected3NodeSpecMultiJvmNode2 extends DownAllIndirectlyConnected3NodeSpec
-class DownAllIndirectlyConnected3NodeSpecMultiJvmNode3 extends DownAllIndirectlyConnected3NodeSpec
+class DownAllWhenIndirectlyConnected3NodeSpecMultiJvmNode1 extends DownAllWhenIndirectlyConnected3NodeSpec
+class DownAllWhenIndirectlyConnected3NodeSpecMultiJvmNode2 extends DownAllWhenIndirectlyConnected3NodeSpec
+class DownAllWhenIndirectlyConnected3NodeSpecMultiJvmNode3 extends DownAllWhenIndirectlyConnected3NodeSpec
 
-class DownAllIndirectlyConnected3NodeSpec extends MultiNodeClusterSpec(DownAllIndirectlyConnected3NodeSpec) {
-  import DownAllIndirectlyConnected3NodeSpec._
+class DownAllWhenIndirectlyConnected3NodeSpec extends MultiNodeClusterSpec(DownAllWhenIndirectlyConnected3NodeSpec) {
+  import DownAllWhenIndirectlyConnected3NodeSpec._
 
   "A 3-node cluster with down-all-when-indirectly-connected=on" should {
     "down all when two unreachable but can talk via third" in {

…ownAllIndirectlyConnected3NodeSpec.scala …llWhenIndirectlyConnected3NodeSpec.scalaakka-cluster/src/multi-jvm/scala/akka/cluster/sbr/DownAllIndirectlyConnected3NodeSpec.scala renamed to akka-cluster/src/multi-jvm/scala/akka/cluster/sbr/DownAllWhenIndirectlyConnected3NodeSpec.scala akka-cluster/src/multi-jvm/scala/akka/cluster/sbr/DownAllIndirectlyConnected3NodeSpec.scala renamed to akka-cluster/src/multi-jvm/scala/akka/cluster/sbr/DownAllWhenIndirectlyConnected3NodeSpec.scala

@@ -0,0 +1,107 @@
+/*
+ * Copyright (C) 2020-2025 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.cluster.sbr
+
+import scala.concurrent.duration._
+
+import com.typesafe.config.ConfigFactory
+
+import akka.cluster.Cluster
+import akka.cluster.MemberStatus
+import akka.cluster.MultiNodeClusterSpec
+import akka.remote.testkit.Direction
+import akka.remote.testkit.MultiNodeConfig
+
+object DownAllWhenIndirectlyConnected5NodeSpec extends MultiNodeConfig {
+  val node1 = role("node1")
+  val node2 = role("node2")
+  val node3 = role("node3")
+  val node4 = role("node4")
+  val node5 = role("node5")
+
+  commonConfig(ConfigFactory.parseString("""
+    akka {
+      loglevel = INFO
+      cluster {
+        downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
+        split-brain-resolver.active-strategy = keep-majority
+        split-brain-resolver.stable-after = 6s
+        # default down-all-when-indirectly-connected = 0.5 (50% threshold)
+
+        run-coordinated-shutdown-when-down = off
+      }
+
+      actor.provider = cluster
+
+      test.filter-leeway = 10s
+    }
+  """))
+
+  testTransport(on = true)
+}
+
+class DownAllWhenIndirectlyConnected5NodeSpecMultiJvmNode1 extends DownAllWhenIndirectlyConnected5NodeSpec
+class DownAllWhenIndirectlyConnected5NodeSpecMultiJvmNode2 extends DownAllWhenIndirectlyConnected5NodeSpec
+class DownAllWhenIndirectlyConnected5NodeSpecMultiJvmNode3 extends DownAllWhenIndirectlyConnected5NodeSpec
+class DownAllWhenIndirectlyConnected5NodeSpecMultiJvmNode4 extends DownAllWhenIndirectlyConnected5NodeSpec
+class DownAllWhenIndirectlyConnected5NodeSpecMultiJvmNode5 extends DownAllWhenIndirectlyConnected5NodeSpec
+
+class DownAllWhenIndirectlyConnected5NodeSpec extends MultiNodeClusterSpec(DownAllWhenIndirectlyConnected5NodeSpec) {
+  import DownAllWhenIndirectlyConnected5NodeSpec._
+
+  "A 5-node cluster with down-all-when-indirectly-connected threshold" should {
+    "down all when indirectly connected cycle involves 3 of 5 nodes (60% > 50% threshold)" in {
+      // This test creates an indirect connection cycle involving nodes 1, 2, 3:
+      //   node1 <-> node2: blocked
+      //   node2 <-> node3: blocked
+      //   node3 <-> node1: blocked
+      // Nodes 4 and 5 can still reach everyone, so gossip flows via them.
+      //
+      // This creates a cycle in the unreachability graph where nodes 1, 2, 3 are both
+      // observers and subjects. The DownIndirectlyConnected decision would down all 3,
+      // which is 60% of the cluster - exceeding the 50% threshold.
+      // Therefore, SBR should down ALL nodes instead.
+
+      val cluster = Cluster(system)
+
+      runOn(node1) {
+        cluster.join(cluster.selfAddress)
+      }
+      enterBarrier("node1 joined")
+      runOn(node2, node3, node4, node5) {
+        cluster.join(node(node1).address)
+      }
+      within(10.seconds) {
+        awaitAssert {
+          cluster.state.members.size should ===(5)
+          cluster.state.members.foreach {
+            _.status should ===(MemberStatus.Up)
+          }
+        }
+      }
+      enterBarrier("Cluster formed")
+
+      // Create indirect connection cycle: node1 <-> node2 <-> node3 <-> node1
+      // All connections between these three nodes are blocked, but they can still
+      // communicate via node4 and node5
+      runOn(node1) {
+        testConductor.blackhole(node1, node2, Direction.Both).await
+        testConductor.blackhole(node2, node3, Direction.Both).await
+        testConductor.blackhole(node3, node1, Direction.Both).await
+      }
+      enterBarrier("blackholed-indirectly-connected-cycle")
+
+      // All nodes should be downed because the indirectly connected set (3 nodes)
+      // exceeds the 50% threshold
+      runOn(node1, node2, node3, node4, node5) {
+        awaitCond(cluster.isTerminated, max = 15.seconds)
+      }
+
+      enterBarrier("done")
+    }
+
+  }
+
+}

akka-cluster/src/multi-jvm/scala/akka/cluster/sbr/DownAllWhenIndirectlyConnected5NodeSpec.scala

@@ -26,6 +26,7 @@ object IndirectlyConnected3NodeSpec extends MultiNodeConfig {
         downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
         split-brain-resolver.active-strategy = keep-majority
         split-brain-resolver.stable-after = 6s
+        split-brain-resolver.down-all-when-indirectly-connected = off
 
         run-coordinated-shutdown-when-down = off
       }

akka-cluster/src/multi-jvm/scala/akka/cluster/sbr/IndirectlyConnected3NodeSpec.scala

@@ -28,6 +28,7 @@ object IndirectlyConnected5NodeSpec extends MultiNodeConfig {
         downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
         split-brain-resolver.active-strategy = keep-majority
         split-brain-resolver.stable-after = 6s
+        split-brain-resolver.down-all-when-indirectly-connected = off
 
         run-coordinated-shutdown-when-down = off
       }