Kotlin协程重复测试导致方法调用顺序异常问题求助

我开发了一套基于Kotlin协程的设备指令序列执行代码。单元测试单次运行正常，但用@RepeatedTest(30)重复执行时，会随机出现调用顺序混乱的失败：有时连续10次成功后连续10次失败，有时从第5-7次重复开始失败，偶尔30次全部通过。

日志排查发现，失败场景下待执行的Step列表顺序异常（CommandStep被集中排列），调用publisher.publishChangeState的协程索引增量在成功和失败时存在差异。试过Debug（未复现）、替换forEach为while循环、调整协程执行方式等，问题仍未解决，请求协助排查原因。

测试代码 #

@RepeatedTest(30)
fun id_7_doubleDelayedScenarioStepTest() = runBlocking {
    val testId = "ID_7"
    //Init - 创建调用场景模板
    val inputTemplateId = scenarioService.createTemplate(
        ScenarioTemplateDto(0, "delayTestScenario", 11, null, testUtils.prepareDoubleDelaySteps(testId), false)
    ).id

    //Mock
    coEvery { publisher.publishChangeState(any()) } returns Unit
    mockkStatic(LocalDateTime::class)
    coEvery { LocalDateTime.now() } returns LocalDateTime.MAX

    //预期输出
    val expectedRequestFirstStep = testUtils.generateOneTimeReq(testId, 1)
    val expectedDelayStart = testUtils.generateIncomingRequest(testId, 2)
    val expectedDelayEnd = testUtils.generateLastRequest(testId, 3)
    val expectedRequestMiddleStep = testUtils.generateOneTimeReq(testId, 4)
    val expectedDelayStart2 = testUtils.generateIncomingRequest(testId, 5)
    val expectedDelayEnd2 = testUtils.generateLastRequest(testId, 6)
    val expectedRequestLastStep = testUtils.generateOneTimeReq(testId, 7)

    //执行
    scenarioService.startScenario(inputTemplateId)

    delay(2200) //等待后台协程执行完成

    //验证
    coVerifySequence {
        publisher.publishChangeState(expectedRequestFirstStep)
        publisher.publishChangeState(expectedDelayStart)
        publisher.publishChangeState(expectedDelayEnd)
        publisher.publishChangeState(expectedRequestMiddleStep)
        publisher.publishChangeState(expectedDelayStart2)
        publisher.publishChangeState(expectedDelayEnd2)
        publisher.publishChangeState(expectedRequestLastStep)
    }
    coVerify(exactly = 7) { publisher.publishChangeState(any()) }
}

生产代码 #

监听器中的协程启动 #

coroutineScope.launch { scenarioExecuteService.processInstance(scenario) }

场景执行逻辑 #

//处理场景的函数
private suspend fun processStepsWithoutInit(steps: MutableList<Step>, instance: Scenario) {
    steps.forEach { step ->
        when (step) {
            is CommandStep -> processCommandStep(step, instance)
            is DelayedStep -> processDelayedStep(step, instance)
            is ToggledStep -> processToggledStep(step, instance)
            else -> throw IllegalArgumentException("Unknown step type: $step.")
        }
    }
    removeLiveInstance(instance.templateId)
}
//处理命令步骤
private fun processCommandStep(step: CommandStep, instance: Scenario) {
    changeStateProduction.sendChangeState(
        step.priority, instance.id, step.commands.map { it.toDto() }, ONE_TIME_REQ
    )
}

//处理延迟步骤
private suspend fun processDelayedStep(step: DelayedStep, instance: Scenario) {
    changeStateProduction.sendChangeState(
        step.priority, instance.id, step.startCommands.map { it.toDto() }, INCOMING_REQ
    )
    executeDelay(step, instance)
}

private suspend fun executeDelay(step: DelayedStep, instance: Scenario) {
    addStoppedInstance(instance, step, coroutineContext[Job]!!)

    delay(step.delay * MILLIS_TO_SECONDS)

    removeStoppedInstance(instance.templateId)
    changeStateProduction.sendChangeState(
        step.priority, instance.id, step.endCommands.map { it.toDto() }, LAST_REQ
    )
}

发布者调用 #

override fun sendChangeState(
    priority: Long,
    scenarioId: Long,
    commandDto: List<CommandDto>,
    incomingRequest: ChangeStateIncomingRequest
) {
    publisher.publishChangeState(ChangeStateDto(priority, scenarioId, commandDto, incomingRequest, LocalDateTime.now()))
}

Kafka发布逻辑 #

fun publishChangeState(dto: ChangeStateDto){
    kafkaSender.sendAsync(ChangeStateTopicIdentifier, dto)
}

失败测试的断言错误（关键部分） #

java.lang.AssertionError: Verification failed: calls are not exactly matching verification sequence
Matchers:
+ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step1, params={})], incomingRequest=ONE_TIME_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
+ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step2, params={})], incomingRequest=INCOMING_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
+ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step3, params={})], incomingRequest=LAST_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step4, params={})], incomingRequest=ONE_TIME_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
+ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step5, params={})], incomingRequest=INCOMING_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
+ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step6, params={})], incomingRequest=LAST_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
ObjectPublisher(objectPublisher bean#1).publishChangeState(eq(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step7, params={})], incomingRequest=ONE_TIME_REQ, timestamp=+999999999-12-31T23:59:59.999999999))))
Calls:
1) +ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step1, params={})], incomingRequest=ONE_TIME_REQ, timestamp=+999999999-12-31T23:59:59.999999999))
2) ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step4, params={})], incomingRequest=ONE_TIME_REQ, timestamp=+999999999-12-31T23:59:59.999999999))
3) ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step7, params={})], incomingRequest=ONE_TIME_REQ, timestamp=+999999999-12-31T23:59:59.999999999))
4) +ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step2, params={})], incomingRequest=INCOMING_REQ, timestamp=+999999999-12-31T23:59:59.999999999))
5) +ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step3, params={})], incomingRequest=LAST_REQ, timestamp=+999999999-12-31T23:59:59.999999999))
6) +ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step5, params={})], incomingRequest=INCOMING_REQ, timestamp=+999999999-12-31T23:59:59.999999999))
7) +ObjectPublisher(objectPublisher bean#1).publishChangeState(ChangeStateDto(priority=1, scenarioId=0, commandDto=[CommandDto(deviceId=1, key=ID_7_step6, params={})], incomingRequest=LAST_REQ, timestamp=+999999999-12-31T23:59:59.999999999))

额外发现 #

• 未Mock时间时，失败场景下调用顺序异常，成功场景顺序正常
• 打印Step列表发现，失败时列表顺序异常（CommandStep集中排列）
• 成功测试时协程索引增量为10，失败时为8

请问这可能是Spring缓存、代码优化或协程使用错误导致的吗？

内容的提问来源于stack exchange，提问作者Jiří Bönsch