如何在Lagom(Scala)集成测试中正确触发事件处理器？

我来帮你排查这个问题——Lagom里切换到JDBC替代Cassandra后，集成测试中事件处理器只初始化不响应事件的情况，通常和测试配置冲突、读侧组件加载不全或者异步处理的等待逻辑有关，咱们一步步来解决：

1. 修正测试环境的Cassandra配置冲突 #

你在测试初始化里用了ServiceTest.defaultSetup.withCassandra(true)，但你的服务已经完全切换到JDBC持久化，这个配置会让Lagom启动Cassandra相关组件，反而会和JDBC读侧的初始化逻辑产生冲突。

解决方案：禁用Cassandra，同时配置测试用的JDBC内存数据库（比如H2）：

override def beforeAll: Unit = {
  val testSetup = ServiceTest.defaultSetup
    .withCassandra(false) // 禁用Cassandra，避免干扰JDBC组件
    .configure(
      "jdbc.default.driver" -> "org.h2.Driver",
      "jdbc.default.url" -> "jdbc:h2:mem:auth-test;DB_CLOSE_DELAY=-1",
      "jdbc.default.user" -> "sa",
      "jdbc.default.password" -> ""
    )
  server = ServiceTest.startServer(testSetup) { ctx => new ServiceTestApplication(ctx) }
  authService = server.serviceClient.implement[AuthService]
}

2. 确认测试应用类继承正确的父类 #

务必保证你的ServiceTestApplication是直接继承自AuthApplication的，否则测试环境不会加载你配置的JDBC组件和事件处理器：

class ServiceTestApplication(context: LagomApplicationContext) extends AuthApplication(context)

3. 检查事件处理器的核心实现是否合规 #

你的DeviceEventProcessor需要满足Lagom JDBC读侧的基本要求，不然会无法正确监听事件：

• 用@ReadSideProcessor注解标记，指定唯一的读侧ID
• 正确重写aggregateTag方法，返回对应实体的事件标签
• 在buildHandler中正确绑定事件与处理逻辑

示例正确的处理器结构：

import com.lightbend.lagom.scaladsl.persistence.jdbc.JdbcReadSide
import com.lightbend.lagom.scaladsl.persistence.{AggregateTag, ReadSideProcessor}

@ReadSideProcessor(readSideId = "device-event-processor")
class DeviceEventProcessor(readSide: JdbcReadSide) extends ReadSideProcessor[DeviceEvent] {
  // 返回DeviceEntity对应的事件标签
  override def aggregateTag: AggregateTag[DeviceEvent] = DeviceEvent.Tag

  override def buildHandler(): ReadSideProcessor.ReadSideHandler[DeviceEvent] = {
    readSide.builder[DeviceEvent]("device-event-offset")
      .setGlobalPrepare(_ => /* 这里可以加表初始化逻辑，比如创建设备表 */)
      .register[DeviceCreated]((event, ctx) => /* 处理设备创建事件的JDBC操作 */)
      .register[DeviceUpdated]((event, ctx) => /* 处理设备更新事件的JDBC操作 */)
      .build()
  }
}

4. 测试中加入异步等待逻辑 #

Lagom的事件处理是异步执行的，发送命令到实体后，事件不会立即被处理器处理。在集成测试中，你需要用eventually来等待读侧处理完成，再做断言：

import org.scalatest.concurrent.Eventually
import scala.concurrent.duration._

class AuthServiceIntegrationTest extends WordSpec with Matchers with Eventually with BeforeAndAfterAll {
  // ... 之前的初始化代码 ...

  "Device service" should {
    "process device created event correctly" in {
      val testDeviceId = "test-device-001"
      authService.createDevice(testDeviceId).invoke().futureValue

      // 等待事件处理器完成异步处理，最多等待5秒
      eventually(timeout(5.seconds)) {
        // 查询JDBC数据库，验证设备记录是否已被处理器写入
        val savedDevice = tokenRepository.findDevice(testDeviceId).futureValue
        savedDevice shouldBe defined
      }
    }
  }
}

5. 验证JDBC偏移量表的初始化 #

Lagom的JDBC读侧依赖read_side_offset表来跟踪事件处理的进度，确保测试数据库自动创建了这个表：

• 可以在H2的JDBC URL中加入INIT=RUNSCRIPT FROM 'classpath:init_tables.sql'，提前执行包含偏移量表创建语句的脚本
• 或者在事件处理器的setGlobalPrepare方法中添加表创建逻辑

内容的提问来源于stack exchange，提问作者idoshamun