akka.pattern.AskTimeoutException 在持久化事件时抛出

Question

我刚开始使用 Lagom 和 Akka。我遵循 Domain Modelling with Akka Persistence Typed

中描述的设计

我正在尝试创建实体的全新实例 (EntityState)。但是事件没有持续存在，我收到以下错误：

00:54:27.862 [error] com.example.impl.entity.EntityClass [persistencePhase=running-cmd, akkaAddress=akka://XXX@127.0.0.1:60685, akkaSource=akka://XXX/system/sharding/StateClass/186/ID1, sourceActorSystem=XXX, persistenceId=StateClass|ID1] - Supervisor StopSupervisor saw failure: null
java.lang.NullPointerException: null
    at akka.persistence.typed.javadsl.EventSourcedBehavior.$anonfun$apply(EventSourcedBehavior.scala:195)
    at akka.persistence.typed.internal.Running$RunningState.applyEvent(Running.scala:78)
    at akka.persistence.typed.internal.Running$HandlingCommands.applyEffects(Running.scala:153)
    at akka.persistence.typed.internal.Running$HandlingCommands.onCommand(Running.scala:123)
    at akka.persistence.typed.internal.Running$HandlingCommands.onMessage(Running.scala:105)
    at akka.persistence.typed.internal.Running$HandlingCommands.onMessage(Running.scala:100)
    at akka.actor.typed.scaladsl.AbstractBehavior.receive(AbstractBehavior.scala:83)

我有一个 Create 命令，它会调用 onCreate()，并最终尝试保留一个 EntityCreated 事件。

服务实现方法

@Override
public ServiceCall<CreateMessage, StateView> createState(){
    return message ->
            entityRef(message.getName())
                    .<EntityClass.Accepted>ask(replyTo -> new EntityClass.Create(message, replyTo), askTimeout)
                    .thenApply(accepted -> toStateView(accepted.getSummary()));
}

命令处理程序：

private ReplyEffect<Event, StateClass> onCreate(StateClass state, Create cmd) {
    return Effect()
            .persist(new EntityCreated(cmd.getDetails().getName(), Instant.now()))
            .thenReply(cmd.replyTo, e -> new Accepted(EntityClass.toSummary(e)));
}

我可以确认以下内容：

persist()
Cassandra 中不存在该事件

感谢您的帮助。提前致谢！

Answer 1

看来异常的真正原因是因为我应该添加处理事件的逻辑如下：在 helloEvents() 中，我需要添加类似于以下的逻辑：

if (eventAndOffset.first() instanceof HelloEvent.GreetingMessageChanged) {
    HelloEvent.GreetingMessageChanged messageChanged = (HelloEvent.GreetingMessageChanged) eventAndOffset.first();
    eventToPublish = new GreetingMessageChanged(messageChanged.getName(), messageChanged.getMessage());
}

此外，在聚合的 eventHandler() 中，我需要添加类似于以下的逻辑：

builder.forAnyState()
    .onEvent(GreetingMessageChanged.class, (state, evt) ->
        // We simply update the current state to use the greeting message from
        // the event.
        state.withMessage(evt.message)
);

akka.pattern.AskTimeoutException 在持久化事件时抛出

akka.pattern.AskTimeoutException is thrown while persisting an event

java

cassandra

akka

lagom