From 84de9b2dfaf38711873bf1d9bddb3cbcbb16ffb1 Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 13:55:18 +0800 Subject: [PATCH] test: stabilize gb32960 durable ack boundary --- .../Gb32960ChannelHandlerAckBoundaryTest.java | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java index ca4cae75..fe83a700 100644 --- a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java +++ b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java @@ -42,9 +42,8 @@ class Gb32960ChannelHandlerAckBoundaryTest { dispatch.awaitPublishCount(1); dispatch.completeDurability(); - channel.runPendingTasks(); - ByteBuf ack = channel.readOutbound(); + ByteBuf ack = readOutboundAfterRunningTasks(channel); assertThat(ack).isNotNull(); assertThat(ack.getUnsignedByte(2)).isEqualTo((short) 0x02); assertThat(ack.getUnsignedByte(3)).isEqualTo((short) 0x01); @@ -85,10 +84,9 @@ class Gb32960ChannelHandlerAckBoundaryTest { assertThat((Object) channel.readOutbound()).isNull(); dispatch.completeDurability(0); - channel.runPendingTasks(); - ByteBuf firstAck = channel.readOutbound(); - ByteBuf secondAck = channel.readOutbound(); + ByteBuf firstAck = readOutboundAfterRunningTasks(channel); + ByteBuf secondAck = readOutboundAfterRunningTasks(channel); assertThat(firstAck).isNotNull(); assertThat(secondAck).isNotNull(); assertThat(firstAck.getUnsignedByte(2)).isEqualTo((short) 0x02); @@ -99,6 +97,19 @@ class Gb32960ChannelHandlerAckBoundaryTest { } } + private static ByteBuf readOutboundAfterRunningTasks(EmbeddedChannel channel) { + long deadline = System.currentTimeMillis() + 3000; + while (System.currentTimeMillis() < deadline) { + channel.runPendingTasks(); + ByteBuf outbound = channel.readOutbound(); + if (outbound != null) { + return outbound; + } + Thread.onSpinWait(); + } + return null; + } + private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) { return new Gb32960ChannelHandler( decoder(),