pjfanning commented on code in PR #3704:
URL: https://github.com/apache/pekko/pull/3704#discussion_r4371801039
##########
distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala:
##########
@@ -2489,6 +2496,8 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
// replace, it's possible that the ordering is changed since it based on
MemberStatus
leader = leader.filterNot(_.uniqueAddress == m.uniqueAddress)
leader += m
+ if (m.address != selfAddress)
Review Comment:
This now makes members first seen as `Down` (via `MemberDowned`) known
senders, so a downed node's Read/Write/DeltaPropagation gets merged until
`MemberRemoved` arrives. In a split-brain scenario, a Replicator that only
learns of node X through `MemberDowned` previously ignored X's writes; now it
applies writes from a node the cluster has already decided to remove. Older
Replicators already accept these since `nodes` isn't cleared on Down, so this
extends existing behaviour rather than introducing it — but is including `Down`
intentional? It would be good to state that explicitly (or exclude it).
##########
distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/ReplicatorKnownNodeSpec.scala:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.cluster.ddata
+
+import org.apache.pekko
+import pekko.actor.Address
+import pekko.cluster.Cluster
+import pekko.cluster.ClusterEvent.MemberDowned
+import pekko.cluster.ClusterEvent.MemberEvent
+import pekko.cluster.ClusterEvent.MemberExited
+import pekko.cluster.ClusterEvent.MemberLeft
+import pekko.cluster.ClusterEvent.MemberPreparingForShutdown
+import pekko.cluster.ClusterEvent.MemberReadyForShutdown
+import pekko.cluster.ClusterEvent.MemberRemoved
+import pekko.cluster.ClusterSettings
+import pekko.cluster.Member
+import pekko.cluster.MemberStatus
+import pekko.cluster.UniqueAddress
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestProbe
+import pekko.util.Version
+
+import com.typesafe.config.ConfigFactory
+
+class ReplicatorKnownNodeSpec
+ extends PekkoSpec(ConfigFactory.parseString("""
+ pekko.actor.provider = cluster
+ pekko.remote.artery.canonical.port = 0
+ pekko.cluster.roles = [replica]
+ """))
+ with ImplicitSender {
+
+ import Replicator.Internal
+
+ private val cluster = Cluster(system)
+
+ override protected def atStartup(): Unit = {
+ cluster.join(cluster.selfAddress)
+ awaitAssert(cluster.selfMember.status shouldEqual MemberStatus.Up)
+ }
+
+ private lazy val replicator = {
Review Comment:
All cases share one lazy `replicator` and the `ImplicitSender` `testActor`,
so members added in earlier cases stay known, and a late
`WriteAck`/`ReadResult` from one case could be consumed by the next. Running a
single case with `-z` or reordering changes the replicator's state (e.g. the
role and incarnation cases run with five other members already known). A fresh
replicator (and probe) per case would make these independent.
##########
distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/ReplicatorKnownNodeSpec.scala:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.cluster.ddata
+
+import org.apache.pekko
+import pekko.actor.Address
+import pekko.cluster.Cluster
+import pekko.cluster.ClusterEvent.MemberDowned
+import pekko.cluster.ClusterEvent.MemberEvent
+import pekko.cluster.ClusterEvent.MemberExited
+import pekko.cluster.ClusterEvent.MemberLeft
+import pekko.cluster.ClusterEvent.MemberPreparingForShutdown
+import pekko.cluster.ClusterEvent.MemberReadyForShutdown
+import pekko.cluster.ClusterEvent.MemberRemoved
+import pekko.cluster.ClusterSettings
+import pekko.cluster.Member
+import pekko.cluster.MemberStatus
+import pekko.cluster.UniqueAddress
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestProbe
+import pekko.util.Version
+
+import com.typesafe.config.ConfigFactory
+
+class ReplicatorKnownNodeSpec
+ extends PekkoSpec(ConfigFactory.parseString("""
+ pekko.actor.provider = cluster
+ pekko.remote.artery.canonical.port = 0
+ pekko.cluster.roles = [replica]
+ """))
+ with ImplicitSender {
+
+ import Replicator.Internal
+
+ private val cluster = Cluster(system)
+
+ override protected def atStartup(): Unit = {
+ cluster.join(cluster.selfAddress)
+ awaitAssert(cluster.selfMember.status shouldEqual MemberStatus.Up)
+ }
+
+ private lazy val replicator = {
+ val ref =
system.actorOf(Replicator.props(ReplicatorSettings(system).withRole("replica")),
"replicator1")
+ val probe = TestProbe()
+ probe.send(ref, Replicator.GetReplicaCount)
+ probe.expectMsg(Replicator.ReplicaCount(1))
+ ref
+ }
+
+ private def memberOn(port: Int): Member =
+ Member(
+ UniqueAddress(Address("pekko", system.name, "127.0.0.1", port),
port.toLong),
+ Set("replica", ClusterSettings.DcRolePrefix +
cluster.settings.SelfDataCenter),
+ Version.Zero)
+
+ private def writeFrom(member: Member, keyId: String): Unit = {
+ val counter = GCounter.empty.increment(member.uniqueAddress, 1)
+ replicator ! Internal.Write(keyId, Internal.DataEnvelope(counter),
Some(member.uniqueAddress))
+ expectMsg(Internal.WriteAck)
+ replicator ! Internal.Read(keyId, Some(member.uniqueAddress))
+ expectMsgType[Internal.ReadResult].envelope.map(_.data) shouldEqual
Some(counter)
+ }
+
+ private def rejectWriteFrom(member: Member, keyId: String): Unit = {
+ val key = GCounterKey(keyId)
+ replicator ! Internal.Write(
+ key.id,
+ Internal.DataEnvelope(GCounter.empty.increment(member.uniqueAddress, 1)),
+ Some(member.uniqueAddress))
+ // A subsequent local read is a barrier: the ignored write must neither
reply nor store data.
+ replicator ! Replicator.Get(key, Replicator.ReadLocal)
+ expectMsg(Replicator.NotFound(key, None))
+ }
+
+ "Replicator" should {
+
+ val events: Seq[MemberEvent] = Seq(
+ MemberLeft(memberOn(12345).copy(MemberStatus.Leaving)),
+
MemberExited(memberOn(12346).copy(MemberStatus.Leaving).copy(MemberStatus.Exiting)),
+ MemberDowned(memberOn(12347).copy(MemberStatus.Down)),
+
MemberPreparingForShutdown(memberOn(12348).copy(MemberStatus.Up).copy(MemberStatus.PreparingForShutdown)),
+ MemberReadyForShutdown(
+
memberOn(12349).copy(MemberStatus.Up).copy(MemberStatus.PreparingForShutdown).copy(
+ MemberStatus.ReadyForShutdown)))
+
+ events.foreach { event =>
+ s"accept Read and Write from a member first observed in
${event.member.status} status" in {
+ replicator ! event
+ writeFrom(event.member, s"known-${event.member.status}")
+ // Recognizing a sender does not add a departing node to the
consistency quorum.
+ replicator ! Replicator.GetReplicaCount
+ expectMsg(Replicator.ReplicaCount(1))
+ }
+ }
+
+ "reject a member after removal from either additional known-node set" in {
Review Comment:
The removal test only covers `MemberLeft` and `MemberExited`. It would be
worth also asserting that `Down`, `PreparingForShutdown` and `ReadyForShutdown`
members are rejected after `MemberRemoved`, since those go through
`otherStatusNodes`; otherwise a regression in clearing that set would leave
downed members known forever with no failing test.
##########
distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala:
##########
@@ -1400,11 +1400,17 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
// cluster exiting nodes, doesn't contain selfAddress
var exitingNodes: immutable.SortedSet[UniqueAddress] =
immutable.SortedSet.empty
+ // Members observed in Leaving, Down, PreparingForShutdown or
ReadyForShutdown status,
+ // excluding selfAddress. They may not have been observed as Up before, but
remain known
+ // senders of Read/Write/DeltaPropagation messages until removed.
+ private var otherStatusNodes: immutable.SortedSet[UniqueAddress] =
immutable.SortedSet.empty
+
// up and weaklyUp nodes, doesn't contain joining and not selfAddress
private def allNodes: immutable.SortedSet[UniqueAddress] =
nodes.union(weaklyUpNodes)
private def isKnownNode(node: UniqueAddress): Boolean =
- nodes(node) || weaklyUpNodes(node) || joiningNodes(node) ||
selfUniqueAddress == node
+ nodes(node) || weaklyUpNodes(node) || joiningNodes(node) ||
exitingNodes(node) || otherStatusNodes(
Review Comment:
`isKnownNode` runs on every Read, Write and DeltaPropagation and now does up
to five `SortedSet` (O(log n)) lookups before the self check.
`otherStatusNodes` is only used for membership tests so it doesn't need to be
sorted; a single `HashSet` of known nodes (see the comment above) would make
this one O(1) lookup.
##########
distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala:
##########
@@ -1400,11 +1400,17 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
// cluster exiting nodes, doesn't contain selfAddress
var exitingNodes: immutable.SortedSet[UniqueAddress] =
immutable.SortedSet.empty
+ // Members observed in Leaving, Down, PreparingForShutdown or
ReadyForShutdown status,
+ // excluding selfAddress. They may not have been observed as Up before, but
remain known
+ // senders of Read/Write/DeltaPropagation messages until removed.
+ private var otherStatusNodes: immutable.SortedSet[UniqueAddress] =
immutable.SortedSet.empty
Review Comment:
The known-sender rule is now spread over six sets (`nodes`, `weaklyUpNodes`,
`joiningNodes`, `exitingNodes`, `otherStatusNodes`, self). That fragmentation
is the root cause of #3689: if a handler forgets to update a set, or a new
`MemberStatus` is added, senders are silently dropped again. Consider a single
"members not yet removed" set, populated from every role-matching `MemberEvent`
and cleared on `MemberRemoved`, used only by `isKnownNode`.
##########
docs/src/main/paradox/distributed-data.md:
##########
@@ -38,6 +38,12 @@ nodes or the majority of nodes. The `WeaklyUp` node is not
counted
as part of the cluster. So 3 nodes + 5 `WeaklyUp` is essentially a
3 node cluster as far as consistent actions are concerned.
+Replication reads, writes and deltas from known members are accepted until
those members are removed,
Review Comment:
"accepted until those members are removed" overstates it for deltas:
`receiveDeltaPropagation` also drops deltas when `isNodeRemoved` is true, which
includes keys whose pruning state already contains the sender, even before
`MemberRemoved`. Maybe soften the wording. The same text is duplicated in
`typed/distributed-data.md`, so both copies need the change.
##########
distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/ReplicatorKnownNodeSpec.scala:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.cluster.ddata
+
+import org.apache.pekko
+import pekko.actor.Address
+import pekko.cluster.Cluster
+import pekko.cluster.ClusterEvent.MemberDowned
+import pekko.cluster.ClusterEvent.MemberEvent
+import pekko.cluster.ClusterEvent.MemberExited
+import pekko.cluster.ClusterEvent.MemberLeft
+import pekko.cluster.ClusterEvent.MemberPreparingForShutdown
+import pekko.cluster.ClusterEvent.MemberReadyForShutdown
+import pekko.cluster.ClusterEvent.MemberRemoved
+import pekko.cluster.ClusterSettings
+import pekko.cluster.Member
+import pekko.cluster.MemberStatus
+import pekko.cluster.UniqueAddress
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestProbe
+import pekko.util.Version
+
+import com.typesafe.config.ConfigFactory
+
+class ReplicatorKnownNodeSpec
+ extends PekkoSpec(ConfigFactory.parseString("""
+ pekko.actor.provider = cluster
+ pekko.remote.artery.canonical.port = 0
+ pekko.cluster.roles = [replica]
+ """))
+ with ImplicitSender {
+
+ import Replicator.Internal
+
+ private val cluster = Cluster(system)
+
+ override protected def atStartup(): Unit = {
+ cluster.join(cluster.selfAddress)
+ awaitAssert(cluster.selfMember.status shouldEqual MemberStatus.Up)
+ }
+
+ private lazy val replicator = {
+ val ref =
system.actorOf(Replicator.props(ReplicatorSettings(system).withRole("replica")),
"replicator1")
+ val probe = TestProbe()
+ probe.send(ref, Replicator.GetReplicaCount)
+ probe.expectMsg(Replicator.ReplicaCount(1))
+ ref
+ }
+
+ private def memberOn(port: Int): Member =
+ Member(
+ UniqueAddress(Address("pekko", system.name, "127.0.0.1", port),
port.toLong),
+ Set("replica", ClusterSettings.DcRolePrefix +
cluster.settings.SelfDataCenter),
+ Version.Zero)
+
+ private def writeFrom(member: Member, keyId: String): Unit = {
+ val counter = GCounter.empty.increment(member.uniqueAddress, 1)
+ replicator ! Internal.Write(keyId, Internal.DataEnvelope(counter),
Some(member.uniqueAddress))
+ expectMsg(Internal.WriteAck)
+ replicator ! Internal.Read(keyId, Some(member.uniqueAddress))
+ expectMsgType[Internal.ReadResult].envelope.map(_.data) shouldEqual
Some(counter)
+ }
+
+ private def rejectWriteFrom(member: Member, keyId: String): Unit = {
+ val key = GCounterKey(keyId)
+ replicator ! Internal.Write(
+ key.id,
+ Internal.DataEnvelope(GCounter.empty.increment(member.uniqueAddress, 1)),
+ Some(member.uniqueAddress))
+ // A subsequent local read is a barrier: the ignored write must neither
reply nor store data.
+ replicator ! Replicator.Get(key, Replicator.ReadLocal)
+ expectMsg(Replicator.NotFound(key, None))
+ }
+
+ "Replicator" should {
+
+ val events: Seq[MemberEvent] = Seq(
+ MemberLeft(memberOn(12345).copy(MemberStatus.Leaving)),
+
MemberExited(memberOn(12346).copy(MemberStatus.Leaving).copy(MemberStatus.Exiting)),
+ MemberDowned(memberOn(12347).copy(MemberStatus.Down)),
+
MemberPreparingForShutdown(memberOn(12348).copy(MemberStatus.Up).copy(MemberStatus.PreparingForShutdown)),
+ MemberReadyForShutdown(
+
memberOn(12349).copy(MemberStatus.Up).copy(MemberStatus.PreparingForShutdown).copy(
+ MemberStatus.ReadyForShutdown)))
+
+ events.foreach { event =>
Review Comment:
These cases cover only Read and Write, but the PR description and both docs
pages say DeltaPropagation from these members is also accepted. Please add a
`DeltaPropagation` case so a regression in that path is caught.
--
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]
