PubSub DeadLetters issues

Lagom Version: 1.3

Attempting to do: Notify all actors in the cluster that an event has been received from a message queue, this prompts them all to update in memory caches.

Issue encountered: When publishing my akka message, my code on the subscription never seems to fire. though i receive the following:

Message [java.lang.String] from Actor[akka://my-impl-application/deadLetters] to Actor[akka://my-impl-application/system/distributedPubSubMediator#855479769] was not delivered. [2] dead letters encountered. This logging can be turned off or adjusted with configuration settings ‘akka.log-dead-letters’ and ‘akka.log-dead-letters-during-shutdown’.

My Investigation: I have tried to find examples and found I was getting similar issues even using example code such as https://github.com/knoldus/lagom-pub-sub.g8

I’ve noted almost all examples seems to suggest subscription be done async to publish logic (in the example the subscription is only done when a stream is opened), which doesn’t suit my needs. An example of the overall logic of my code is below (this isn’t the code, it’s a nice simplification):

public class MyService {
    @Inject
    //KafkaService is just an instance of com.lightbend.lagom.javadsl.api.Service with a single Topic
    public MyService(KafkaService service, PubSubRegistry pubSubRegistry) {
        PubSubRef<String> pubSubRef = pubSubRegistry.refFor(TopicId.of(String.class, "my_topic"));
        service.myTopic().subscribe()
            .atLeastOnce(Flow.<String>create().fromFunction(msg -> {
                pubSubRef.publish(msg);
                return Done.getInstance();
            });
        pubSubRef.subscriber().subscribe().map(msg -> {
            System.out.println(msg);
            return msg;
        });
    }
}

application.conf

play.akka.actor-system = "my-impl-application"
akka {
  actor {
    provider = "cluster"
  }
  remote {
    log-remote-lifecycle-events = off
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2552
    }
  }
}
akka.cluster.seed-nodes = [
  "akka.tcp://my-impl-application@127.0.0.1:2552"]

Expected: When an actor in my cluster receives a kafka event, an akka msg is published which ALL other actors consume and then do something.

Current Workaround: Currently as per https://doc.akka.io/docs/akka/2.4/java/distributed-pub-sub.html I’m just building my own akka pubsub structure, though I would much prefer to use the built in Lagom feature

Where are you running your subscription graph ? Seems like you just build it.

How do you mean? As in it looks like in my code I’m just building the service but never user it.
I should probably have clarified I am injecting the service into a controller for the purposes of being created.

@Maraket
When you do

pubSubRegistry.refFor(TopicId.of(String.class, “my_topic”));

this ill return a PubSubRef[T] rigth ? After that you subscribe to the topic

pubSubRef.subscriber().subscribe()

this ill return a Source[T,Mat] rigth ? Then you need to run the source graph. If you can ignore you sink do that:

.runWith(Sink.ignore)(Keep.left)

This ill execute your graph until that you ill not be abble to read any messages because you are just building the graph

How would that be in Java for lagom 1.3? As far as I can tell Akka 2.4 requires runWith with 2 parameters not 1.

@Maraket
With akka 2.4.20 you can use that =)

The second parameter is an Generic type for the materializer, you can use the runWIth function as i show before

That was it, thank you VictorBLS, it works thanks for all the help

1 Like