"Address already in use" error from ScalaTestWithActorTestKit

Akka
1

with

extends ScalaTestWithActorTestKit(ConfigFactory.parseString("""
    // akka.persistence.testkit.events.serialize = off
    // akka {
    //   remote {
    //       artery.enabled = on
    //       artery.transport = aeron-udp
    //       artery.canonical.port = 0
    //       artery.canonical.hostname = localhost
    //   }
    // }
    """))

which is just an empty config, I got the following error in my test using ScalaTestWithActorTestKit

[2021-02-08 23:12:45,054] [ERROR] [akka.remote.artery.aeron.ArteryAeronUdpTransport] [] [my-app-akka.actor.default-dispatcher-6] - Aeron error: 1 observations from 02/08/2021 23:12:45.043 to 02/08/2021 23:12:45.043 for:
 io.aeron.exceptions.AeronException: ERROR - channel error - Address already in use (at java.base/sun.nio.ch.Net.bind0(Native Method)): aeron:udp?endpoint=127.0.0.1:2552
        at io.aeron.driver.media.UdpChannelTransport.openDatagramChannel(UdpChannelTransport.java:167)
        at io.aeron.driver.media.ReceiveChannelEndpoint.openChannel(ReceiveChannelEndpoint.java:240)
        at io.aeron.driver.Receiver.onRegisterReceiveChannelEndpoint(Receiver.java:156)
        at io.aeron.driver.ReceiverProxy.lambda$registerReceiveChannelEndpoint$5(ReceiverProxy.java:126)
        at org.agrona.concurrent.OneToOneConcurrentArrayQueue.drain(OneToOneConcurrentArrayQueue.java:116)
        at io.aeron.driver.Receiver.doWork(Receiver.java:77)
        at org.agrona.concurrent.CompositeAgent.doWork(CompositeAgent.java:114)
        at org.agrona.concurrent.AgentRunner.doDutyCycle(AgentRunner.java:291)
        at org.agrona.concurrent.AgentRunner.run(AgentRunner.java:164)
        at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.net.BindException: Address already in use
        at java.base/sun.nio.ch.Net.bind0(Native Method)
        at java.base/sun.nio.ch.Net.bind(Net.java:455)
        at java.base/sun.nio.ch.DatagramChannelImpl.bindInternal(DatagramChannelImpl.java:814)
        at java.base/sun.nio.ch.DatagramChannelImpl.bind(DatagramChannelImpl.java:785)
        at io.aeron.driver.media.UdpChannelTransport.openDatagramChannel(UdpChannelTransport.java:127)
        ... 9 more

and

akka.remote.RemoteTransportException was thrown inside "my-test" when, construction cannot continue: "Inbound Aeron channel is in errored state. See Aeron logs for details."

I was expecting it to work out of the box. what config do I need to pass to ScalaTestWithActorTestKit?

2

If I don’t parse any config, I get

[2021-02-08 23:25:06,162] [ERROR] [akka.cluster.typed.internal.receptionist.ClusterReceptionist] [] [MySpec-akka.actor.internal-dispatcher-20] - cannot create top-level actor from the outside on ActorSystem with custom user guardian {akkaAddress=akka://MySpec@127.0.0.1:59749, akkaSource=akka://MySpec/system/clusterReceptionist, sourceActorSystem=MySpec}
akka.actor.ActorInitializationException: akka://MySpec/system/clusterReceptionist/replicator: exception during creation
        at akka.actor.ActorInitializationException$.apply(Actor.scala:196)
        at akka.actor.ActorCell.create(ActorCell.scala:661)
        at akka.actor.ActorCell.invokeAll$1(ActorCell.scala:513)
        at akka.actor.ActorCell.systemInvoke(ActorCell.scala:535)
        at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:295)
        at akka.dispatch.Mailbox.run(Mailbox.scala:230)
        at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
        at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
        at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
        at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
        at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
        at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
Caused by: java.lang.UnsupportedOperationException: cannot create top-level actor from the outside on ActorSystem with custom user guardian
        at akka.actor.ActorSystemImpl.actorOf(ActorSystem.scala:901)
        at my.app.util.serialization.MyOwnMessageSerializer.<init>(MyOwnMessageSerializer.scala:34)
        at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)
        at akka.actor.ReflectiveDynamicAccess.$anonfun$createInstanceFor$1(ReflectiveDynamicAccess.scala:40)
        at scala.util.Try$.apply(Try.scala:210)
        at akka.actor.ReflectiveDynamicAccess.createInstanceFor(ReflectiveDynamicAccess.scala:35)
        at akka.actor.ReflectiveDynamicAccess.$anonfun$createInstanceFor$5(ReflectiveDynamicAccess.scala:48)
        at scala.util.Success.flatMap(Try.scala:258)
        at akka.actor.ReflectiveDynamicAccess.createInstanceFor(ReflectiveDynamicAccess.scala:47)
        at akka.serialization.Serialization.serializerOf(Serialization.scala:390)
        at akka.serialization.Serialization.$anonfun$serializers$2(Serialization.scala:424)
        at scala.collection.Iterator$$anon$9.next(Iterator.scala:575)
        at scala.collection.immutable.HashMapBuilder.addAll(HashMap.scala:2360)
        at scala.collection.immutable.HashMap$.from(HashMap.scala:2182)
        at scala.collection.immutable.HashMap$.from(HashMap.scala:2158)
        at scala.collection.MapOps$WithFilter.map(Map.scala:348)
        at akka.serialization.Serialization.<init>(Serialization.scala:424)
        at akka.serialization.SerializationExtension$.createExtension(SerializationExtension.scala:18)
        at akka.serialization.SerializationExtension$.createExtension(SerializationExtension.scala:14)
        at akka.actor.ActorSystemImpl.registerExtension(ActorSystem.scala:1158)
        at akka.actor.ExtensionId.apply(Extension.scala:78)
        at akka.actor.ExtensionId.apply$(Extension.scala:77)
        at akka.serialization.SerializationExtension$.apply(SerializationExtension.scala:14)
        at akka.remote.artery.Encoder$$anon$1.serialization(Codecs.scala:87)
        at akka.remote.artery.Encoder$$anon$1.onPush(Codecs.scala:129)
        at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:541)
        at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:423)
        at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:625)
        at akka.stream.impl.fusing.GraphInterpreterShell$AsyncInput.execute(ActorGraphInterpreter.scala:502)
        at akka.stream.impl.fusing.GraphInterpreterShell.processEvent(ActorGraphInterpreter.scala:600)
        at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:773)
        at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:788)
        at akka.actor.Actor.aroundReceive(Actor.scala:537)
        at akka.actor.Actor.aroundReceive$(Actor.scala:535)
        at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:691)
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:577)
        at akka.actor.ActorCell.invoke(ActorCell.scala:547)
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
        at akka.dispatch.Mailbox.run(Mailbox.scala:231)
        ... 6 common frames omitted

which indicate that it tries to initialize MyOwnMessageSerializer.
whose implementation looks like

class MyOwnMessageSerializer(system: ExtendedActorSystem) extends SerializerWithStringManifest {
...
private val distributedRegistry =
    system.actorOf(
      PropsAdapter(
        DistributedRegistry(
          "AvroMessageSerializer",
          localSchemaRegistry)))
...
}
3

I think there are two facts/issues involved.

  1. it seems the testkit needs a real binding, which makes tests less reproducible
  2. there is no typed actor system API for user-defined serializers

I’d like to understand why it tries to load my serializer (via reflection?) at all when I don’t pass any config to ScalaTestWithActorTestKit.

note: it used to work well a few versions back. so I assume some changes in 2.6.12 or 2.6.11 brought it up.

4

The solution is to replace

extends ScalaTestWithActorTestKit

with

extends ScalaTestWithActorTestKit("""
        akka.persistence.testkit.events.serialize = off
      """)

I think it used to be off in earlier versions and now on by default.

5

The cause is that I have implicit val system = ActorSystem("my-app") in my test.
It is fixed by replacing it with implicit val system = testKit.internalSystem.toClassic