This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new e2169744d9 fix: handle Artery uid collisions gracefully and make
system uids more unique (#3652)
e2169744d9 is described below
commit e2169744d9a84351234565924b01ad3ec210ec54
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 16:13:48 2026 +0100
fix: handle Artery uid collisions gracefully and make system uids more
unique (#3652)
* fix: handle Artery uid collision more gracefully
Motivation:
When two remote systems with the same ActorSystem uid connect to a node,
AssociationRegistry.setUID threw a plain IllegalArgumentException from
completeHandshake. That escaped the inbound handshake stage, failed the
inbound stream and could disturb the healthy association that already
owned the uid.
Modification:
- Introduce internal Association.UidCollisionException (NoStackTrace) and
throw it from AssociationRegistry.setUID on collision.
- ArteryTransport.completeHandshake logs a warning, quarantines the
colliding peer uid and returns a failed Future instead of throwing.
- InboundHandshake only replies with HandshakeRsp when completing the
handshake succeeded, and keeps pulling either way.
- ActorSystemImpl reads an internal, test-only `pekko.test-only-uid`
setting (when present) so uid collisions can be tested. It is not in
reference.conf and the uid is random when it is absent.
- Add RemoteConnectionSpec tests for collisions when connecting to and
from two systems with the same uid; give each test its own echo actor
name.
Result:
A uid collision is rejected for the new peer only; the existing
association keeps working in both directions. No wire format change, and
classic remoting is unaffected.
Tests:
- Not run locally - left to CI; scalafmt applied
References:
None - robustness improvement for Artery handshake
* fix: generate a more unique Artery system uid
Motivation:
The actor system uid was a plain random Long picked when the ActorSystem is
created. With Artery, two systems ending up with the same uid make
associations fail, so the uid should be harder to collide across nodes.
Modification:
- Add an internal `systemUid` to `ActorRefProvider` and `RemoteTransport`;
`ExtendedActorSystem.uid` now delegates to the provider and throws an
IllegalStateException if it is read before the provider is initialized.
- `LocalActorRefProvider` and classic `Remoting` keep a random uid.
- `ArteryTransport` generates the uid when it starts, mixing the bind host,
port and several timestamps (one taken early in `RemoteActorRefProvider`).
The internal `pekko.test-only-uid` setting moves here.
- `Slf4jLogger` resolves the uid lazily and omits the `pekkoUid` MDC value
until it is available.
- MiMa excludes for the new abstract methods on the @DoNotInherit
`ActorRefProvider` and the internal `RemoteTransport`.
Result:
Artery system uids are much less likely to collide. The uid is still a
Long on the wire, so mixed-version clusters interoperate.
Tests:
- Not run locally - left to CI; scalafmt applied
References:
None - Artery uid uniqueness
* test: assert the Artery uid collision warning and use probe timeouts
Motivation:
Review of #3652 noted that EventFilter[UidCollisionException]() does not
assert a collision: it matches Error events with that exception and allows
any number of them, while the collision is logged as a Warning without the
exception. The collision tests also still used explicit 500 ms deadlines,
which the Akka 2.9.0 follow-up (akka-core#32177) replaced with the
configured
probe timeout, and the adapted files kept their older Lightbend notices.
Modification:
- Subscribe a probe to Warning events and expect a "UID collision" warning
after each collision step. Handshake retries can log the collision more
than
once, so the tests expect at least one.
- Use the default probe timeout for the eight round trips in the two
collision
tests; the retry-polling tests keep their deadlines.
- Update the Lightbend notices of the files adapted from Akka 2.9.0 to the
2023 source notices.
Result:
Removing the collision warning makes both collision tests fail.
Tests:
sbt "remote/testOnly org.apache.pekko.remote.artery.RemoteConnectionSpec
org.apache.pekko.remote.artery.SystemUidSpec"
(9 passed; both collision tests fail with the warning removed)
sbt "actor/headerCheckAll" "remote/headerCheckAll" "slf4j/headerCheckAll"
References:
https://github.com/apache/pekko/pull/3652
---
.../2.0.x.backwards.excludes/system-uid.excludes | 19 ++
.../org/apache/pekko/actor/ActorRefProvider.scala | 12 +-
.../scala/org/apache/pekko/actor/ActorSystem.scala | 17 +-
.../2.0.x.backwards.excludes/system-uid.excludes | 19 ++
.../pekko/remote/RemoteActorRefProvider.scala | 9 +-
.../org/apache/pekko/remote/RemoteTransport.scala | 6 +-
.../scala/org/apache/pekko/remote/Remoting.scala | 4 +
.../pekko/remote/artery/ArteryTransport.scala | 44 ++++-
.../apache/pekko/remote/artery/Association.scala | 7 +-
.../org/apache/pekko/remote/artery/Handshake.scala | 18 +-
.../pekko/remote/artery/RemoteConnectionSpec.scala | 199 ++++++++++++++++++++-
.../apache/pekko/remote/artery/SystemUidSpec.scala | 66 +++++++
.../org/apache/pekko/event/slf4j/Slf4jLogger.scala | 20 ++-
13 files changed, 410 insertions(+), 30 deletions(-)
diff --git
a/actor/src/main/mima-filters/2.0.x.backwards.excludes/system-uid.excludes
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/system-uid.excludes
new file mode 100644
index 0000000000..ff6e0ba2a9
--- /dev/null
+++ b/actor/src/main/mima-filters/2.0.x.backwards.excludes/system-uid.excludes
@@ -0,0 +1,19 @@
+# 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.
+
+# Additional method in ActorRefProvider, which is @DoNotInherit
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.actor.ActorRefProvider.systemUid")
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala
b/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala
index 8655e38bf8..eb836b0e10 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala
@@ -8,11 +8,12 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.actor
+import java.util.concurrent.ThreadLocalRandom
import java.util.concurrent.atomic.AtomicLong
import scala.annotation.implicitNotFound
@@ -178,6 +179,12 @@ import pekko.util.OptionVal
@InternalApi
private[pekko] def addressString: String
+ /**
+ * INTERNAL API: the uid of the actor system, see `ExtendedActorSystem.uid`
+ */
+ @InternalApi
+ private[pekko] def systemUid: Long
+
}
/**
@@ -789,4 +796,7 @@ private[pekko] class LocalActorRefProvider private[pekko] (
addr
}
}
+
+ override private[pekko] val systemUid: Long =
+ ThreadLocalRandom.current.nextLong()
}
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
index e88f364cbe..7857c2fdc6 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.actor
@@ -834,8 +834,6 @@ private[pekko] class ActorSystemImpl(
setup: ActorSystemSetup)
extends ExtendedActorSystem {
- val uid: Long = ThreadLocalRandom.current.nextLong()
-
if (!name.matches("""^[a-zA-Z0-9][a-zA-Z0-9-_]*$"""))
throw new IllegalArgumentException(
"invalid ActorSystem name [" + name +
@@ -861,6 +859,19 @@ private[pekko] class ActorSystemImpl(
new Settings(classLoader, config, name, setup)
}
+ override def uid: Long = {
+ try {
+ provider.systemUid
+ } catch {
+ case NonFatal(exc) =>
+ // could be NPE in RemoteActorRefProvider if the transport is not
initialized yet
+ throw new IllegalStateException(
+ "uid accessed before provider has been initialized. " +
+ "This is a bug, please report at
https://github.com/apache/pekko/issues";,
+ exc)
+ }
+ }
+
protected def uncaughtExceptionHandler: Thread.UncaughtExceptionHandler =
new Thread.UncaughtExceptionHandler() {
def uncaughtException(thread: Thread, cause: Throwable): Unit = {
diff --git
a/remote/src/main/mima-filters/2.0.x.backwards.excludes/system-uid.excludes
b/remote/src/main/mima-filters/2.0.x.backwards.excludes/system-uid.excludes
new file mode 100644
index 0000000000..8298a50725
--- /dev/null
+++ b/remote/src/main/mima-filters/2.0.x.backwards.excludes/system-uid.excludes
@@ -0,0 +1,19 @@
+# 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.
+
+# Additional method in internal RemoteTransport
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.remote.RemoteTransport.systemUid")
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/RemoteActorRefProvider.scala
b/remote/src/main/scala/org/apache/pekko/remote/RemoteActorRefProvider.scala
index 098e9ac00e..d287424545 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/RemoteActorRefProvider.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/RemoteActorRefProvider.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.remote
@@ -167,6 +167,10 @@ private[pekko] class RemoteActorRefProvider(
extends ActorRefProvider {
import RemoteActorRefProvider._
+ // One part of the system uid generation. Taking this early to allow for
some execution randomness until later
+ // when actually generating the uid.
+ private[pekko] val systemUidTimestamp1: Long = System.nanoTime()
+
val remoteSettings: RemoteSettings = new RemoteSettings(settings.config)
private[pekko] final val hasClusterOrUseUnsafe =
@@ -698,6 +702,9 @@ private[pekko] class RemoteActorRefProvider(
local.addressString
}
}
+
+ override private[pekko] def systemUid: Long =
+ transport.systemUid
}
private[pekko] trait RemoteRef extends ActorRefScope {
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/RemoteTransport.scala
b/remote/src/main/scala/org/apache/pekko/remote/RemoteTransport.scala
index 2977c33466..aa547c71bb 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/RemoteTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/RemoteTransport.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.remote
@@ -110,4 +110,8 @@ private[pekko] abstract class RemoteTransport(val system:
ExtendedActorSystem, v
@InternalStableApi
def quarantine(address: Address, uid: Option[Long], reason: String): Unit
+ /**
+ * The uid of the actor system, see `ExtendedActorSystem.uid`.
+ */
+ def systemUid: Long
}
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
b/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
index 76f4d6446e..b585a47f3e 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
@@ -16,6 +16,7 @@ package org.apache.pekko.remote
import java.net.URLEncoder
import java.nio.charset.StandardCharsets
import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ThreadLocalRandom
import java.util.concurrent.TimeoutException
import scala.annotation.nowarn
@@ -164,6 +165,9 @@ private[remote] class Remoting(_system:
ExtendedActorSystem, _provider: RemoteAc
override def localAddressForRemote(remote: Address): Address =
Remoting.localAddressForRemote(transportMapping, remote)
+ // classic remoting only uses the lower 32 bits of the uid, see
AddressUidExtension
+ override val systemUid: Long = ThreadLocalRandom.current.nextLong()
+
val log: LoggingAdapter = Logging(system.eventStream, classOf[Remoting])
val eventPublisher = new EventPublisher(system, log,
RemoteLifecycleEventsLogLevel)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
index ed59f3b43d..cb4df8adf9 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2016-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2016-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.remote.artery
@@ -39,12 +39,12 @@ import pekko.event.Logging
import pekko.event.MarkerLoggingAdapter
import pekko.pattern.after
import pekko.protobufv3.internal.CodedInputStream
-import pekko.remote.AddressUidExtension
import pekko.remote.ContainerFormats
import pekko.remote.RemoteActorRef
import pekko.remote.RemoteActorRefProvider
import pekko.remote.RemoteTransport
import pekko.remote.UniqueAddress
+import pekko.remote.artery.Association.UidCollisionException
import pekko.remote.artery.Decoder.InboundCompressionAccess
import pekko.remote.artery.Encoder.OutboundCompressionAccess
import pekko.remote.artery.InboundControlJunction.ControlMessageObserver
@@ -329,6 +329,11 @@ private[remote] abstract class ArteryTransport(_system:
ExtendedActorSystem, _pr
override def addresses: Set[Address] = _addresses
override def localAddressForRemote(remote: Address): Address = defaultAddress
+ /**
+ * Must not be accessed before `start()`, will throw NullPointerException
otherwise.
+ */
+ override def systemUid: Long = _localAddress.uid
+
protected val killSwitch: SharedKillSwitch =
KillSwitches.shared("transportKillSwitch")
// keyed by the streamId
@@ -406,14 +411,20 @@ private[remote] abstract class ArteryTransport(_system:
ExtendedActorSystem, _pr
val (port, boundPort) = bindInboundStreams()
+ val uid =
+ if (provider.settings.config.hasPath("pekko.test-only-uid"))
+ provider.settings.config.getLong("pekko.test-only-uid") // to be able
to test uid collisions
+ else
+ generateSystemUid(settings.Bind.Hostname, port,
provider.systemUidTimestamp1)
+
_localAddress = UniqueAddress(
Address(provider.remoteSettings.ProtocolName, system.name,
settings.Canonical.Hostname, port),
- AddressUidExtension(system).longAddressUid)
+ uid)
_addresses = Set(_localAddress.address)
_bindAddress = UniqueAddress(
Address(provider.remoteSettings.ProtocolName, system.name,
settings.Bind.Hostname, boundPort),
- AddressUidExtension(system).longAddressUid)
+ uid)
flightRecorder.transportUniqueAddressSet(_localAddress)
@@ -774,7 +785,12 @@ private[remote] abstract class ArteryTransport(_system:
ExtendedActorSystem, _pr
try {
associationRegistry.setUID(peer).completeHandshake(peer)
} catch {
- case ShuttingDown => Future.successful(Done) // silence it
+ case ShuttingDown => Future.successful(Done) // silence it
+ case exc: UidCollisionException =>
+ log.warning(exc.getMessage)
+ // quarantine will not do much since handshake not completed, but for
good measures
+ quarantine(peer.address, Some(peer.uid), exc.getMessage, harmless =
false)
+ Future.failed(exc)
}
}
@@ -1103,4 +1119,22 @@ private[remote] object ArteryTransport {
case _ => "message"
}
+ /**
+ * Generates the system uid from the bind hostname and port in combination
with timestamps,
+ * which makes collisions between systems less likely than with a plain
random number.
+ */
+ def generateSystemUid(hostname: String, port: Int, timestamp1: Long): Long =
{
+ var hash1 = 23
+ hash1 = 31 * hash1 + hostname.hashCode
+ hash1 = 31 * hash1 + java.lang.Long.hashCode(timestamp1)
+ hash1 = 31 * hash1 + java.lang.Long.hashCode(System.currentTimeMillis())
+
+ var hash2 = 23
+ hash2 = 31 * hash2 + port
+ hash2 = 31 * hash2 + java.lang.Long.hashCode(System.nanoTime())
+ hash2 = 31 * hash2 + java.lang.Long.hashCode(System.currentTimeMillis())
+
+ (hash1.toLong << 32) | (hash2 & 0xFFFFFFFFL)
+ }
+
}
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
index 19f6e3a025..62702ebfc3 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2016-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2016-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.remote.artery
@@ -50,6 +50,7 @@ import pekko.remote.RemoteLogMarker
import pekko.remote.UniqueAddress
import pekko.remote.artery.ArteryTransport.AeronTerminated
import pekko.remote.artery.ArteryTransport.ShuttingDown
+import pekko.remote.artery.Association.UidCollisionException
import pekko.remote.artery.Encoder.OutboundCompressionAccess
import pekko.remote.artery.InboundControlJunction.ControlMessageSubject
import pekko.remote.artery.OutboundControlJunction.OutboundControlIngress
@@ -130,6 +131,8 @@ private[remote] object Association {
streamKillSwitch: OptionVal[SharedKillSwitch],
completed: Future[Done],
stopping: OptionVal[StopSignal])
+
+ final class UidCollisionException(message: String) extends
IllegalArgumentException(message) with NoStackTrace
}
/**
@@ -1190,7 +1193,7 @@ private[remote] class
AssociationRegistry(createAssociation: Address => Associat
a
else
// make sure we don't overwrite same UID with different association
- throw new IllegalArgumentException(s"UID collision old [$previous]
new [$a]")
+ throw new UidCollisionException(s"UID collision old [$previous] new
[$a]")
case _ =>
// update associationsByUid Map with the uid -> association
val newMap = currentMap.updated(peer.uid, a)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala
index 0c4c3c3deb..a338eb00db 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2016-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2016-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.remote.artery
@@ -258,7 +258,7 @@ private[remote] class InboundHandshake(inboundContext:
InboundContext, inControl
// that the other system is alive.
inboundContext.association(from.address).associationState.lastUsedTimestamp.set(System.nanoTime())
- after(inboundContext.completeHandshake(from)) { () =>
+ after(inboundContext.completeHandshake(from)) { _ =>
pull(in)
}
case _ =>
@@ -281,8 +281,9 @@ private[remote] class InboundHandshake(inboundContext:
InboundContext, inControl
private def onHandshakeReq(from: UniqueAddress, to: Address): Unit = {
if (to == inboundContext.localAddress.address) {
- after(inboundContext.completeHandshake(from)) { () =>
- inboundContext.sendControl(from.address,
HandshakeRsp(inboundContext.localAddress))
+ after(inboundContext.completeHandshake(from)) { success =>
+ if (success)
+ inboundContext.sendControl(from.address,
HandshakeRsp(inboundContext.localAddress))
pull(in)
}
} else {
@@ -300,15 +301,16 @@ private[remote] class InboundHandshake(inboundContext:
InboundContext, inControl
}
}
- private def after(first: Future[Done])(thenInside: () => Unit): Unit = {
+ private def after(first: Future[Done])(thenInside: Boolean => Unit):
Unit = {
first.value match {
- case Some(_) =>
+ case Some(result) =>
// This in the normal case (all but the first). The future will be
completed
// because handshake was already completed. Note that we send
those HandshakeReq
// periodically.
- thenInside()
+ thenInside(result.isSuccess)
case None =>
- first.onComplete(_ =>
runInStage.invoke(thenInside))(ExecutionContext.parasitic)
+ first.onComplete(result => runInStage.invoke(() =>
thenInside(result.isSuccess)))(
+ ExecutionContext.parasitic)
}
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
index 997bb106ca..aa46f16877 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.remote.artery
@@ -17,6 +17,11 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.actor.ActorSystem
+import pekko.actor.ExtendedActorSystem
+import pekko.event.Logging
+import pekko.remote.RARP
+import pekko.remote.artery.AssociationState.UidKnown
+import pekko.remote.artery.AssociationState.UidUnknown
import pekko.testkit.{ EventFilter, ImplicitSender, TestActors, TestEvent,
TestProbe }
class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-closed-for = 5s") with
ImplicitSender {
@@ -29,16 +34,195 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
EventFilter.warning(pattern = "received dead letter.*")))
}
+ private def warningsProbe(system: ActorSystem): TestProbe = {
+ val probe = new TestProbe(system)
+ system.eventStream.subscribe(probe.ref, classOf[Logging.Warning])
+ probe
+ }
+
+ // handshake retries can log the collision more than once, so expect at
least one warning
+ private def expectCollisionWarning(warnings: TestProbe): Unit =
+ warnings.fishForMessage(hint = "UID collision warning") {
+ case w: Logging.Warning => w.message.toString.startsWith("UID collision
old [")
+ }
+
"Remoting between systems" should {
+ "handle uid collision when connection TO two systems with same uid" in {
+ val localProbe = new TestProbe(localSystem)
+ val localPort =
RARP(localSystem).provider.getDefaultAddress.getPort().get
+
+ val conflictUid = localSystem.asInstanceOf[ExtendedActorSystem].uid + 1
+ val echoName = "echoA"
+
+ val remotePort1 = freePort()
+ val remoteSystem1 =
+ newRemoteSystem(extraConfig = Some(s"""
+ pekko.test-only-uid = $conflictUid
+ pekko.remote.artery.canonical.port=$remotePort1
+ """))
+ val remote1Probe = new TestProbe(remoteSystem1)
+
+ val remotePort2 = freePort()
+ val remoteSystem2 =
+ newRemoteSystem(extraConfig = Some(s"""
+ // same uid as remoteSystem1
+ pekko.test-only-uid = $conflictUid
+ pekko.remote.artery.canonical.port=$remotePort2
+ """))
+ val remote2Probe = new TestProbe(remoteSystem2)
+
+ localProbe.expectNoMessage(2.seconds)
+ localSystem.actorOf(TestActors.echoActorProps, echoName)
+ remoteSystem1.actorOf(TestActors.echoActorProps, echoName)
+ remoteSystem2.actorOf(TestActors.echoActorProps, echoName)
+
+ val selectionFromLocalToRemote1 =
+
localSystem.actorSelection(s"pekko://${remoteSystem1.name}@localhost:$remotePort1/user/$echoName")
+ val selectionFromLocalToRemote2 =
+
localSystem.actorSelection(s"pekko://${remoteSystem2.name}@localhost:$remotePort2/user/$echoName")
+ val selectionFromRemote1ToLocal =
+
remoteSystem1.actorSelection(s"pekko://${localSystem.name}@localhost:$localPort/user/$echoName")
+ val selectionFromRemote2ToLocal =
+
remoteSystem2.actorSelection(s"pekko://${localSystem.name}@localhost:$localPort/user/$echoName")
+
+ selectionFromLocalToRemote1.tell("ping1a", localProbe.ref)
+ localProbe.expectMsg("ping1a")
+
+ selectionFromRemote1ToLocal.tell("ping1b", remote1Probe.ref)
+ remote1Probe.expectMsg("ping1b")
+
+ val collisionWarnings1 = warningsProbe(localSystem)
+ selectionFromLocalToRemote2.tell("ping2a", localProbe.ref)
+ // doesn't get through
+ localProbe.expectNoMessage()
+ expectCollisionWarning(collisionWarnings1)
+ RARP(localSystem).provider.transport
+ .asInstanceOf[ArteryTransport]
+ .association(RARP(remoteSystem2).provider.getDefaultAddress)
+ .associationState
+ .uniqueRemoteAddressState() shouldBe UidUnknown // handshake not
completed
+
+ val collisionWarnings2 = warningsProbe(localSystem)
+ selectionFromRemote2ToLocal.tell("ping2b", remote2Probe.ref)
+ // doesn't get through in other direction
+ remote2Probe.expectNoMessage()
+ expectCollisionWarning(collisionWarnings2)
+
+ RARP(remoteSystem2).provider.transport
+ .asInstanceOf[ArteryTransport]
+ .association(RARP(localSystem).provider.getDefaultAddress)
+ .associationState
+ .uniqueRemoteAddressState() shouldBe UidKnown // handshake was
completed in other direction
+
+ RARP(localSystem).provider.transport
+ .asInstanceOf[ArteryTransport]
+ .association(RARP(remoteSystem1).provider.getDefaultAddress)
+ .associationState
+ .uniqueRemoteAddressState() shouldBe UidKnown // still intact
+
+ // still works
+ selectionFromLocalToRemote1.tell("ping1a again", localProbe.ref)
+ localProbe.expectMsg("ping1a again")
+
+ // still works in other direction
+ selectionFromRemote1ToLocal.tell("ping1b again", remote1Probe.ref)
+ remote1Probe.expectMsg("ping1b again")
+ }
+
+ "handle uid collision when connection FROM two systems with same uid" in {
+ // same kind of test as above, but connection is first established from
the other remote systems
+ val localProbe = new TestProbe(localSystem)
+ val localPort =
RARP(localSystem).provider.getDefaultAddress.getPort().get
+
+ val conflictUid = localSystem.asInstanceOf[ExtendedActorSystem].uid + 2
+ val echoName = "echoB"
+
+ val remotePort1 = freePort()
+ val remoteSystem1 =
+ newRemoteSystem(extraConfig = Some(s"""
+ pekko.test-only-uid = $conflictUid
+ pekko.remote.artery.canonical.port=$remotePort1
+ """))
+ val remote1Probe = new TestProbe(remoteSystem1)
+
+ val remotePort2 = freePort()
+ val remoteSystem2 =
+ newRemoteSystem(extraConfig = Some(s"""
+ // same uid as remoteSystem1
+ pekko.test-only-uid = $conflictUid
+ pekko.remote.artery.canonical.port=$remotePort2
+ """))
+ val remote2Probe = new TestProbe(remoteSystem2)
+
+ localProbe.expectNoMessage(2.seconds)
+ localSystem.actorOf(TestActors.echoActorProps, echoName)
+ remoteSystem1.actorOf(TestActors.echoActorProps, echoName)
+ remoteSystem2.actorOf(TestActors.echoActorProps, echoName)
+
+ val selectionFromLocalToRemote1 =
+
localSystem.actorSelection(s"pekko://${remoteSystem1.name}@localhost:$remotePort1/user/$echoName")
+ val selectionFromLocalToRemote2 =
+
localSystem.actorSelection(s"pekko://${remoteSystem2.name}@localhost:$remotePort2/user/$echoName")
+ val selectionFromRemote1ToLocal =
+
remoteSystem1.actorSelection(s"pekko://${localSystem.name}@localhost:$localPort/user/$echoName")
+ val selectionFromRemote2ToLocal =
+
remoteSystem2.actorSelection(s"pekko://${localSystem.name}@localhost:$localPort/user/$echoName")
+
+ selectionFromRemote1ToLocal.tell("ping1b", remote1Probe.ref)
+ remote1Probe.expectMsg("ping1b")
+
+ selectionFromLocalToRemote1.tell("ping1a", localProbe.ref)
+ localProbe.expectMsg("ping1a")
+
+ val collisionWarnings3 = warningsProbe(localSystem)
+ selectionFromRemote2ToLocal.tell("ping2b", remote2Probe.ref)
+ // doesn't get through
+ localProbe.expectNoMessage()
+ expectCollisionWarning(collisionWarnings3)
+ RARP(localSystem).provider.transport
+ .asInstanceOf[ArteryTransport]
+ .association(RARP(remoteSystem2).provider.getDefaultAddress)
+ .associationState
+ .uniqueRemoteAddressState() shouldBe UidUnknown // handshake not
completed
+
+ val collisionWarnings4 = warningsProbe(localSystem)
+ selectionFromLocalToRemote2.tell("ping2a", localProbe.ref)
+ // doesn't get through in other direction
+ remote2Probe.expectNoMessage()
+ expectCollisionWarning(collisionWarnings4)
+
+ RARP(remoteSystem2).provider.transport
+ .asInstanceOf[ArteryTransport]
+ .association(RARP(localSystem).provider.getDefaultAddress)
+ .associationState
+ .uniqueRemoteAddressState() shouldBe UidKnown // handshake was
completed in other direction
+
+ RARP(localSystem).provider.transport
+ .asInstanceOf[ArteryTransport]
+ .association(RARP(remoteSystem1).provider.getDefaultAddress)
+ .associationState
+ .uniqueRemoteAddressState() shouldBe UidKnown // still intact
+
+ // still works
+ selectionFromRemote1ToLocal.tell("ping1b again", remote1Probe.ref)
+ remote1Probe.expectMsg("ping1b again")
+
+ // still works in other direction
+ selectionFromLocalToRemote1.tell("ping1a again", localProbe.ref)
+ localProbe.expectMsg("ping1a again")
+ }
+
"be able to connect to system even if it's not there at first" in {
muteSystem(localSystem)
val localProbe = new TestProbe(localSystem)
+ val echoName = "echoC"
val remotePort = freePort()
// try to talk to it before it is up
- val selection =
localSystem.actorSelection(s"pekko://$nextGeneratedSystemName@localhost:$remotePort/user/echo")
+ val selection =
+
localSystem.actorSelection(s"pekko://$nextGeneratedSystemName@localhost:$remotePort/user/$echoName")
selection.tell("ping", localProbe.ref)
localProbe.expectNoMessage(1.seconds)
@@ -47,7 +231,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
muteSystem(remoteSystem)
localProbe.expectNoMessage(2.seconds)
- remoteSystem.actorOf(TestActors.echoActorProps, "echo")
+ remoteSystem.actorOf(TestActors.echoActorProps, echoName)
within(5.seconds) {
awaitAssert {
@@ -59,17 +243,19 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
"allow other system to connect even if it's not there at first" in {
val localSystem = newRemoteSystem()
+ val echoName = "echoD"
val localPort = port(localSystem)
muteSystem(localSystem)
val localProbe = new TestProbe(localSystem)
- localSystem.actorOf(TestActors.echoActorProps, "echo")
+ localSystem.actorOf(TestActors.echoActorProps, echoName)
val remotePort = freePort()
// try to talk to remote before it is up
- val selection =
localSystem.actorSelection(s"pekko://$nextGeneratedSystemName@localhost:$remotePort/user/echo")
+ val selection =
+
localSystem.actorSelection(s"pekko://$nextGeneratedSystemName@localhost:$remotePort/user/$echoName")
selection.tell("ping", localProbe.ref)
localProbe.expectNoMessage(1.seconds)
@@ -80,7 +266,8 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
localProbe.expectNoMessage(2.seconds)
val otherProbe = new TestProbe(remoteSystem)
val otherSender = otherProbe.ref
- val thisSelection =
remoteSystem.actorSelection(s"pekko://${localSystem.name}@localhost:$localPort/user/echo")
+ val thisSelection =
+
remoteSystem.actorSelection(s"pekko://${localSystem.name}@localhost:$localPort/user/$echoName")
within(5.seconds) {
awaitAssert {
thisSelection.tell("ping", otherSender)
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/SystemUidSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/SystemUidSpec.scala
new file mode 100644
index 0000000000..d8cbe1e061
--- /dev/null
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/SystemUidSpec.scala
@@ -0,0 +1,66 @@
+/*
+ * 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.remote.artery
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.actor.ExtendedActorSystem
+import pekko.remote.AddressUidExtension
+import pekko.remote.RARP
+import pekko.testkit.TestKit
+
+class SystemUidSpec extends ArteryMultiNodeSpec {
+
+ private def uidOf(sys: ActorSystem): Long =
sys.asInstanceOf[ExtendedActorSystem].uid
+
+ "The system uid with Artery" must {
+
+ "be the uid of the transport local address" in {
+ val transport =
RARP(localSystem).provider.transport.asInstanceOf[ArteryTransport]
+ uidOf(localSystem) should ===(transport.localAddress.uid)
+ uidOf(localSystem) should ===(transport.bindAddress.uid)
+ AddressUidExtension(localSystem).longAddressUid should
===(uidOf(localSystem))
+ }
+
+ "be stable" in {
+ uidOf(localSystem) should ===(uidOf(localSystem))
+ }
+
+ "be different for different systems" in {
+ val otherSystem = newRemoteSystem()
+ uidOf(otherSystem) should !==(uidOf(localSystem))
+ }
+
+ "use the test-only-uid when defined" in {
+ val otherSystem = newRemoteSystem(extraConfig =
Some("pekko.test-only-uid = 4711"))
+ uidOf(otherSystem) should ===(4711L)
+ }
+ }
+
+ "The system uid without remoting" must {
+
+ "be stable" in {
+ val localOnly = ActorSystem("SystemUidSpec-local")
+ try {
+ uidOf(localOnly) should ===(uidOf(localOnly))
+ } finally {
+ TestKit.shutdownActorSystem(localOnly)
+ }
+ }
+ }
+}
diff --git
a/slf4j/src/main/scala/org/apache/pekko/event/slf4j/Slf4jLogger.scala
b/slf4j/src/main/scala/org/apache/pekko/event/slf4j/Slf4jLogger.scala
index c14a078022..43227cbaee 100644
--- a/slf4j/src/main/scala/org/apache/pekko/event/slf4j/Slf4jLogger.scala
+++ b/slf4j/src/main/scala/org/apache/pekko/event/slf4j/Slf4jLogger.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.event.slf4j
@@ -75,7 +75,19 @@ class Slf4jLogger extends Actor with SLF4JLogging with
RequiresMessageQueue[Logg
val mdcPekkoUidAttributeName = "pekkoUid"
private def pekkoAddress =
context.system.asInstanceOf[ExtendedActorSystem].provider.addressString
- private val pekkoUid: String =
context.system.asInstanceOf[ExtendedActorSystem].uid.toString
+ // the uid is not available until the remote transport has been started, so
it is resolved lazily
+ private var _pekkoUid: String = ""
+ private def pekkoUid: String = {
+ if (_pekkoUid.isEmpty) {
+ try {
+ _pekkoUid =
context.system.asInstanceOf[ExtendedActorSystem].uid.toString
+ } catch {
+ case _: IllegalStateException =>
+ // when logging before the uid is generated (transport binding when
remoting is enabled)
+ }
+ }
+ _pekkoUid
+ }
def receive = {
@@ -135,7 +147,9 @@ class Slf4jLogger extends Actor with SLF4JLogging with
RequiresMessageQueue[Logg
MDC.put(mdcPekkoTimestamp, formatTimestamp(logEvent.timestamp))
MDC.put(mdcActorSystemAttributeName, context.system.name)
MDC.put(mdcPekkoAddressAttributeName, pekkoAddress)
- MDC.put(mdcPekkoUidAttributeName, pekkoUid)
+ val uid = pekkoUid
+ if (uid.nonEmpty)
+ MDC.put(mdcPekkoUidAttributeName, uid)
logEvent.mdc.foreach { case (k, v) => MDC.put(k, String.valueOf(v)) }
try logStatement
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
