多集成测试并发时SmallRye InMemoryConnector.clear()失效问题

问题场景 #

同时运行多个发布事件的集成测试时，InMemoryConnector.clear()无法正常清空消息队列，导致当前测试的队列中混入其他测试产生的事件，破坏了测试的独立性。

现有代码 #

测试类 #

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@QuarkusTest
@QuarkusTestResource(
    value = InMemReactiveMessagingLifecycleManager::class, initArgs = [
        ResourceArg(value = "incoming", name = "example-processed"),
        ResourceArg(value = "outgoing", name = "example")
    ]
)
class DemoEventProducerTest {

    @Inject
    lateinit var demoEventProducer: DemoEventProducer

    @Inject
    @Any
    lateinit var inMemoryConnector: InMemoryConnector

    @Test
    fun `publishEvent - valid action - event published`() {
        demoEventProducer.publish()

        val eventQueue = inMemoryConnector.sink<String>("example")
        // 队列中包含了其他测试发布的事件
    }
}

测试资源管理器 #

class InMemReactiveMessagingLifecycleManager : QuarkusTestResourceLifecycleManager {

    private val params: MutableMap<String, String> = HashMap()

    override fun init(params: Map<String, String>?) {
        this.params.putAll(params!!)
    }

    override fun start(): Map<String, String> {
        val env: MutableMap<String, String> = HashMap()
        for ((key, value) in params) {
            when (value) {
                "incoming" -> env.putAll(InMemoryConnector.switchIncomingChannelsToInMemory(key))
                "outgoing" -> env.putAll(InMemoryConnector.switchOutgoingChannelsToInMemory(key))
            }
        }
        return env
    }

    override fun stop() {
        InMemoryConnector.clear()
    }
}

问题原因分析 #

1. @TestInstance(TestInstance.Lifecycle.PER_CLASS)使得测试类仅创建一个实例，InMemoryConnector作为应用级Bean在测试方法间共享。
2. InMemReactiveMessagingLifecycleManager的stop()方法仅在整个测试类执行完毕后才调用，无法在单个测试方法间清理队列。
3. 直接调用全局InMemoryConnector.clear()可能未针对性清理注册的通道，或者通道存在缓存机制导致清理不彻底。

解决方案 #

方案1：在每个测试方法前针对性清理通道 #

放弃全局clear()，直接清理测试中用到的sink和source，确保每个测试执行前队列是空的：

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@QuarkusTest
@QuarkusTestResource(...)
class DemoEventProducerTest {

    // ... 已有的注入代码

    @BeforeEach
    fun clearQueues() {
        // 清理输出通道的队列
        inMemoryConnector.sink<String>("example").clear()
        // 清理输入通道的队列（如果测试中用到的话）
        inMemoryConnector.source<String>("example-processed").clear()
    }

    @Test
    fun `publishEvent - valid action - event published`() {
        demoEventProducer.publish()

        val eventQueue = inMemoryConnector.sink<String>("example")
        // 此时队列仅包含当前测试产生的事件
    }
}

方案2：调整测试实例生命周期（可选） #

如果不需要PER_CLASS生命周期，可以去掉@TestInstance注解（默认是PER_METHOD，每个测试方法创建新的测试实例），配合@BeforeEach清理队列，进一步隔离测试环境。

方案3：使用TestExecutionListener全局控制清理 #

如果需要在多个测试类中复用清理逻辑，可以实现TestExecutionListener来在每个测试方法前后执行清理：

class InMemoryConnectorCleanupListener : TestExecutionListener {
    override fun beforeTestExecution(context: TestExtensionContext?) {
        // 通过Arc容器获取InMemoryConnector实例
        val connector = Arc.container()
            .instance(InMemoryConnector::class.java, Any.Literal.INSTANCE)
            .get()
        
        // 清理指定通道
        connector.sink<String>("example").clear()
        connector.source<String>("example-processed").clear()
    }
}

然后在测试类上添加注解启用该监听器：

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@QuarkusTest
@TestExecutionListeners(
    listeners = [InMemoryConnectorCleanupListener::class],
    mergeMode = TestExecutionListeners.MergeMode.MERGE_WITH_DEFAULTS
)
@QuarkusTestResource(...)
class DemoEventProducerTest {
    // ... 测试代码
}

验证效果 #

执行多个测试方法后，每个测试的eventQueue只会包含当前测试产生的事件，测试独立性得到保证。

内容的提问来源于stack exchange，提问作者Dimas Adriyanto Setyawan