From a80b38bd04bf59e77824c0553266ea770c5dd36b Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 23 Jun 2026 17:27:12 +0800 Subject: [PATCH] fix: enforce gb32960 kafka ack boundary --- .../core/concurrency/DisruptorEventBus.java | 32 ++- .../ingest/core/dispatcher/Dispatcher.java | 29 ++- .../DisruptorEventBusAwaitTest.java | 61 +++++ .../DispatcherDurableAckBoundaryTest.java | 217 ++++++++++++++++++ .../inbound/Gb32960ChannelHandler.java | 78 +++++-- .../Gb32960ChannelHandlerAckBoundaryTest.java | 87 ++++++- 6 files changed, 474 insertions(+), 30 deletions(-) create mode 100644 modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java create mode 100644 modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherDurableAckBoundaryTest.java diff --git a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java index bf586cde..6558bf33 100644 --- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java +++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java @@ -13,8 +13,12 @@ import com.lmax.disruptor.dsl.ProducerType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; import java.util.List; +import java.util.Objects; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicLong; @@ -30,11 +34,13 @@ public final class DisruptorEventBus implements AutoCloseable { private final Disruptor disruptor; private final List sinks; + private final ExecutorService awaitExecutor; private final AtomicLong published = new AtomicLong(); private final AtomicLong failed = new AtomicLong(); public DisruptorEventBus(int ringBufferSize, String waitStrategyName, List sinks) { this.sinks = List.copyOf(sinks); + this.awaitExecutor = Executors.newVirtualThreadPerTaskExecutor(); ThreadFactory tf = Thread.ofVirtual().name("ingest-bus-", 0).factory(); this.disruptor = new Disruptor<>( VehicleEventSlot::new, @@ -58,21 +64,31 @@ public final class DisruptorEventBus implements AutoCloseable { published.incrementAndGet(); } - public CompletableFuture publishAndAwait(VehicleEvent event) { + public CompletableFuture publishAndAwait(VehicleEvent event, String requiredSinkName) { + String requiredSink = Objects.requireNonNull(requiredSinkName, "requiredSinkName"); published.incrementAndGet(); - List> futures = sinks.stream() - .filter(sink -> sink.accepts(event)) - .map(sink -> publishToSinkAndTrack(sink, event)) - .toList(); - if (futures.isEmpty()) { - return CompletableFuture.completedFuture(null); + List> requiredFutures = new ArrayList<>(); + for (EventSink sink : sinks) { + if (!sink.accepts(event)) continue; + + CompletableFuture future = CompletableFuture + .supplyAsync(() -> publishToSinkAndTrack(sink, event), awaitExecutor) + .thenCompose(f -> f); + if (requiredSink.equals(sink.name())) { + requiredFutures.add(future); + } } - return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)); + if (requiredFutures.isEmpty()) { + return CompletableFuture.failedFuture(new IllegalStateException( + "required sink '" + requiredSink + "' did not accept event " + event.eventId())); + } + return CompletableFuture.allOf(requiredFutures.toArray(CompletableFuture[]::new)); } @Override public void close() { disruptor.shutdown(); + awaitExecutor.shutdown(); log.info("DisruptorEventBus stopped published={} failed={}", published.get(), failed.get()); } diff --git a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java index 02b7e78c..083ee314 100644 --- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java +++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java @@ -35,6 +35,7 @@ public final class Dispatcher { private static final Logger log = LoggerFactory.getLogger(Dispatcher.class); private static final AtomicLong RAW_ARCHIVE_SEQUENCE = new AtomicLong(); + private static final String REQUIRED_DURABLE_SINK = "kafka"; private final HandlerRegistry registry; private final InterceptorChain interceptors; @@ -57,7 +58,7 @@ public final class Dispatcher { public void dispatch(RawFrame frame) { IngestContext ctx = new IngestContext(UUID.randomUUID().toString()); try { - DispatchPlan plan = mapFrameToEvents(frame, ctx); + DispatchPlan plan = mapFrameToEvents(frame, ctx, false); for (VehicleEvent e : plan.events()) { eventBus.publish(e); } @@ -73,15 +74,21 @@ public final class Dispatcher { public CompletableFuture dispatchAndAwait(RawFrame frame) { IngestContext ctx = new IngestContext(UUID.randomUUID().toString()); try { - DispatchPlan plan = mapFrameToEvents(frame, ctx); + DispatchPlan plan = mapFrameToEvents(frame, ctx, true); + if (plan.events().isEmpty() && plan.failure() == null) { + return CompletableFuture.failedFuture(new IllegalStateException( + "awaited dispatch produced no events; required sink '" + REQUIRED_DURABLE_SINK + "' was not reached")); + } CompletableFuture[] futures = plan.events().stream() - .map(eventBus::publishAndAwait) + .map(event -> eventBus.publishAndAwait(event, REQUIRED_DURABLE_SINK)) .toArray(CompletableFuture[]::new); CompletableFuture boundary = CompletableFuture.allOf(futures); if (plan.failure() == null) { return boundary; } return boundary.handle((ignored, publishFailure) -> { + log.error("dispatch failure traceId={}", ctx.traceId(), plan.failure()); + interceptors.onError(plan.failure(), ctx); if (publishFailure != null) { throw new CompletionException(publishFailure); } @@ -102,7 +109,7 @@ public final class Dispatcher { return future.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS); } - private DispatchPlan mapFrameToEvents(RawFrame frame, IngestContext ctx) { + private DispatchPlan mapFrameToEvents(RawFrame frame, IngestContext ctx, boolean awaitDurability) { List out = new ArrayList<>(); // 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit // 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 Sink 消费。 @@ -125,6 +132,11 @@ public final class Dispatcher { for (HandlerDefinition def : handlers) { if (def.asyncBatch() != null) { + if (awaitDurability) { + throw new AwaitedAsyncBatchUnsupportedException( + "@AsyncBatch handler cannot be used with dispatchAndAwait until batch completion is awaited: " + + def.method()); + } batchExecutor.submit(def, frame.payload(), event -> enrichWithRawArchive(event, rawArchive)); continue; } @@ -137,6 +149,9 @@ public final class Dispatcher { } return new DispatchPlan(out, null); } catch (Throwable t) { + if (t instanceof AwaitedAsyncBatchUnsupportedException) { + return new DispatchPlan(List.of(), t); + } return new DispatchPlan(out, t); } } @@ -238,4 +253,10 @@ public final class Dispatcher { private record DispatchPlan(List events, Throwable failure) { } + + private static final class AwaitedAsyncBatchUnsupportedException extends IllegalStateException { + private AwaitedAsyncBatchUnsupportedException(String message) { + super(message); + } + } } diff --git a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java new file mode 100644 index 00000000..2e4fe905 --- /dev/null +++ b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java @@ -0,0 +1,61 @@ +package com.lingniu.ingest.core.concurrency; + +import com.lingniu.ingest.api.ProtocolId; +import com.lingniu.ingest.api.event.VehicleEvent; +import com.lingniu.ingest.api.sink.EventSink; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import static org.assertj.core.api.Assertions.assertThat; + +class DisruptorEventBusAwaitTest { + + @Test + void publishAndAwaitDoesNotInvokeSinkPublishOnCallerThread() throws Exception { + long callerThreadId = Thread.currentThread().threadId(); + CapturingSink sink = new CapturingSink("kafka"); + + try (DisruptorEventBus bus = new DisruptorEventBus(1024, "blocking", java.util.List.of(sink))) { + bus.publishAndAwait(event(), "kafka").get(3, TimeUnit.SECONDS); + } + + assertThat(sink.publishThreadId.get()).isNotEqualTo(callerThreadId); + } + + private static VehicleEvent.Heartbeat event() { + Instant now = Instant.parse("2026-06-23T10:00:00Z"); + return new VehicleEvent.Heartbeat( + "heartbeat-1", + "VIN001", + ProtocolId.GB32960, + now, + now, + "trace-1", + Map.of()); + } + + private static final class CapturingSink implements EventSink { + private final String name; + private final AtomicLong publishThreadId = new AtomicLong(-1); + + private CapturingSink(String name) { + this.name = name; + } + + @Override + public String name() { + return name; + } + + @Override + public CompletableFuture publish(VehicleEvent event) { + publishThreadId.set(Thread.currentThread().threadId()); + return CompletableFuture.completedFuture(null); + } + } +} diff --git a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherDurableAckBoundaryTest.java b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherDurableAckBoundaryTest.java new file mode 100644 index 00000000..25c2c1e6 --- /dev/null +++ b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherDurableAckBoundaryTest.java @@ -0,0 +1,217 @@ +package com.lingniu.ingest.core.dispatcher; + +import com.lingniu.ingest.api.ProtocolId; +import com.lingniu.ingest.api.annotation.AsyncBatch; +import com.lingniu.ingest.api.annotation.MessageMapping; +import com.lingniu.ingest.api.annotation.ProtocolHandler; +import com.lingniu.ingest.api.event.VehicleEvent; +import com.lingniu.ingest.api.pipeline.IngestContext; +import com.lingniu.ingest.api.pipeline.IngestInterceptor; +import com.lingniu.ingest.api.pipeline.RawFrame; +import com.lingniu.ingest.api.sink.EventSink; +import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor; +import com.lingniu.ingest.core.concurrency.DisruptorEventBus; +import com.lingniu.ingest.core.pipeline.InterceptorChain; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class DispatcherDurableAckBoundaryTest { + + @Test + void dispatchAndAwaitFailsWhenNoKafkaSinkAcceptsEvent() throws Exception { + EventSink fileSink = new NamedSink("event-file-store", true); + + try (Harness harness = new Harness(registryWithHeartbeatHandler(), List.of(fileSink), List.of())) { + CompletionStage future = harness.dispatcher.dispatchAndAwait(frame(0x07, "payload")); + + assertThatThrownBy(() -> future.toCompletableFuture().join()) + .isInstanceOf(CompletionException.class) + .hasRootCauseInstanceOf(IllegalStateException.class) + .hasMessageContaining("kafka"); + } + } + + @Test + void dispatchAndAwaitNotifiesInterceptorWhenPlanCapturesFailure() throws Exception { + CapturingInterceptor interceptor = new CapturingInterceptor(); + + try (Harness harness = new Harness(registryWithThrowingHandler(), List.of(new NamedSink("kafka", true)), + List.of(interceptor))) { + CompletionStage future = harness.dispatcher.dispatchAndAwait(frame(0x08, "payload")); + + assertThatThrownBy(() -> future.toCompletableFuture().join()) + .isInstanceOf(CompletionException.class) + .hasRootCauseInstanceOf(IllegalStateException.class) + .rootCause() + .hasMessageContaining("handler failed"); + assertThat(interceptor.error.get()) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Handler threw exception"); + } + } + + @Test + void dispatchAndAwaitFailsFastForAsyncBatchHandler() throws Exception { + NamedSink kafka = new NamedSink("kafka", true); + try (Harness harness = new Harness(registryWithAsyncBatchHandler(), List.of(kafka), List.of())) { + CompletionStage future = harness.dispatcher.dispatchAndAwait( + frame(0x09, "payload", new byte[]{0x23, 0x23, 0x09})); + + assertThatThrownBy(() -> future.toCompletableFuture().join()) + .isInstanceOf(CompletionException.class) + .hasRootCauseInstanceOf(IllegalStateException.class) + .hasMessageContaining("@AsyncBatch"); + assertThat(kafka.published.get()).isZero(); + } + } + + private static HandlerRegistry registryWithHeartbeatHandler() throws NoSuchMethodException { + HeartbeatHandler bean = new HeartbeatHandler(); + Method method = HeartbeatHandler.class.getDeclaredMethod("onHeartbeat", Object.class); + return registry(bean, method, 0x07, method.getAnnotation(AsyncBatch.class)); + } + + private static HandlerRegistry registryWithThrowingHandler() throws NoSuchMethodException { + ThrowingHandler bean = new ThrowingHandler(); + Method method = ThrowingHandler.class.getDeclaredMethod("onFrame", Object.class); + return registry(bean, method, 0x08, method.getAnnotation(AsyncBatch.class)); + } + + private static HandlerRegistry registryWithAsyncBatchHandler() throws NoSuchMethodException { + AsyncHandler bean = new AsyncHandler(); + Method method = AsyncHandler.class.getDeclaredMethod("onBatch", List.class); + return registry(bean, method, 0x09, method.getAnnotation(AsyncBatch.class)); + } + + private static HandlerRegistry registry(Object bean, Method method, int command, AsyncBatch asyncBatch) { + HandlerRegistry registry = new HandlerRegistry(); + registry.register(new HandlerDefinition( + ProtocolId.GB32960, + command, + 0, + method.getName(), + bean, + method, + Object.class, + null, + null, + asyncBatch)); + return registry; + } + + private static RawFrame frame(int command, Object payload) { + return frame(command, payload, null); + } + + private static RawFrame frame(int command, Object payload, byte[] rawBytes) { + return new RawFrame( + ProtocolId.GB32960, + command, + 0, + payload, + rawBytes, + Map.of("vin", "VIN001"), + Instant.parse("2026-06-23T10:00:00Z")); + } + + private static VehicleEvent.Heartbeat event(String id) { + Instant now = Instant.parse("2026-06-23T10:00:00Z"); + return new VehicleEvent.Heartbeat(id, "VIN001", ProtocolId.GB32960, now, now, "trace-" + id, Map.of()); + } + + private static final class Harness implements AutoCloseable { + private final DisruptorEventBus eventBus; + private final AsyncBatchExecutor batchExecutor; + private final Dispatcher dispatcher; + + private Harness(HandlerRegistry registry, List sinks, List interceptors) { + this.eventBus = new DisruptorEventBus(1024, "blocking", sinks); + this.batchExecutor = new AsyncBatchExecutor(eventBus::publish); + this.dispatcher = new Dispatcher( + registry, + new InterceptorChain(interceptors), + new HandlerInvoker(), + eventBus, + batchExecutor); + } + + @Override + public void close() { + batchExecutor.close(); + eventBus.close(); + } + } + + private static final class NamedSink implements EventSink { + private final String name; + private final boolean accepts; + private final AtomicInteger published = new AtomicInteger(); + + private NamedSink(String name, boolean accepts) { + this.name = name; + this.accepts = accepts; + } + + @Override + public String name() { + return name; + } + + @Override + public CompletableFuture publish(VehicleEvent event) { + published.incrementAndGet(); + return CompletableFuture.completedFuture(null); + } + + @Override + public boolean accepts(VehicleEvent event) { + return accepts; + } + } + + private static final class CapturingInterceptor implements IngestInterceptor { + private final AtomicReference error = new AtomicReference<>(); + + @Override + public void onError(Throwable error, IngestContext ctx) { + this.error.set(error); + } + } + + @ProtocolHandler(protocol = ProtocolId.GB32960) + static final class HeartbeatHandler { + @MessageMapping(command = 0x07) + VehicleEvent.Heartbeat onHeartbeat(Object ignored) { + return event("heartbeat-1"); + } + } + + @ProtocolHandler(protocol = ProtocolId.GB32960) + static final class ThrowingHandler { + @MessageMapping(command = 0x08) + VehicleEvent.Heartbeat onFrame(Object ignored) { + throw new IllegalStateException("handler failed"); + } + } + + @ProtocolHandler(protocol = ProtocolId.GB32960) + public static final class AsyncHandler { + @MessageMapping(command = 0x09) + @AsyncBatch(size = 2, waitMs = 10, poolSize = 1) + public List onBatch(List ignored) { + return List.of(event("async-1")); + } + } +} diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java index cd23252b..96b2bcc5 100644 --- a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java @@ -20,6 +20,7 @@ import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.time.Instant; +import java.util.concurrent.CompletableFuture; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -55,6 +56,9 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { public static final AttributeKey PLATFORM_ACCOUNT_ATTR = Gb32960AccessService.PLATFORM_ACCOUNT_ATTR; + private static final AttributeKey> DURABLE_ACK_CHAIN_ATTR = + AttributeKey.valueOf("gb32960.durableAckChain"); + private final Gb32960MessageDecoder decoder; private final Dispatcher dispatcher; private final Gb32960AccessService accessService; @@ -123,7 +127,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { } // 鉴权通过：先完成命令侧状态变更/日志，再统一 dispatch 到通用管线。 - // 需要成功 ACK 的已接收帧，等待 dispatch 的持久化边界完成后再回 ACK。 + // 需要成功 ACK 的已接收帧，等待配置的 Kafka dispatch 边界完成后再回 ACK。 // 平台登入鉴权失败会短路 return，不进 dispatch。 switch (cmd) { case VEHICLE_LOGIN -> logVehicleLogin(ctx, msg); @@ -143,16 +147,62 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { return; } - dispatcher.dispatchAndAwait(rf).whenComplete((ignored, ex) -> - ctx.executor().execute(() -> { - if (ex != null) { - log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}", - addr(ctx), vin, Integer.toHexString(cmd.code()), ex); - ctx.close(); - return; - } - writeAckForCommand(ctx, msg, rawVin); - })); + enqueueDurableAck(ctx, msg, rawVin, vin, cmd, dispatcher.dispatchAndAwait(rf)); + } + + private void enqueueDurableAck(ChannelHandlerContext ctx, + Gb32960Message msg, + byte[] rawVin, + String vin, + CommandType cmd, + CompletableFuture dispatchFuture) { + CompletableFuture previous = ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).get(); + if (previous == null) { + previous = CompletableFuture.completedFuture(null); + } + + CompletableFuture next = previous + .handle((ignored, previousFailure) -> null) + .thenCompose(ignored -> dispatchFuture.handle((ok, failure) -> failure)) + .thenCompose(failure -> runOnEventLoop(ctx, + () -> handleDurableAckResult(ctx, msg, rawVin, vin, cmd, failure))); + ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).set(next); + } + + private CompletableFuture runOnEventLoop(ChannelHandlerContext ctx, Runnable task) { + CompletableFuture done = new CompletableFuture<>(); + Runnable wrapped = () -> { + try { + task.run(); + done.complete(null); + } catch (Throwable t) { + done.completeExceptionally(t); + } + }; + if (ctx.executor().inEventLoop()) { + wrapped.run(); + } else { + ctx.executor().execute(wrapped); + } + return done; + } + + private void handleDurableAckResult(ChannelHandlerContext ctx, + Gb32960Message msg, + byte[] rawVin, + String vin, + CommandType cmd, + Throwable failure) { + if (!ctx.channel().isActive()) { + return; + } + if (failure != null) { + log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}", + addr(ctx), vin, Integer.toHexString(cmd.code()), failure); + ctx.close(); + return; + } + writeAckForCommand(ctx, msg, rawVin); } private RawFrame rawFrame(ChannelHandlerContext ctx, Gb32960Message msg, byte[] frame) { @@ -304,11 +354,11 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { } /** - * 0x02/0x03/0x07 实时 / 补发 / 心跳：记录帧诊断；成功 ACK 在 dispatch 持久化边界后写回。 + * 0x02/0x03/0x07 实时 / 补发 / 心跳：记录帧诊断；成功 ACK 在配置的 Kafka dispatch 边界后写回。 * 部分车端实现严格按规范，没收到应答会停止后续上报或重连。ack-tag 按 cmd.code 区分便于日志筛选。 * - *

注意：应答成功不代表下游已完成持久化，只表示平台已接收并通过 Dispatcher 投递。 - * 生产排查存储问题时，应继续检查 raw archive sink、EventFileStoreSink 和 Kafka sink 的日志。 + *

注意：应答成功表示当前配置的 Kafka sink 已完成该帧映射事件的 dispatch 边界。 + * 生产排查其他存储问题时，应继续检查 raw archive sink、EventFileStoreSink 等可选 sink 的日志。 * *

日志策略（按 (channel, vin, rawTypeSignature) 去重，避免高频帧刷屏）： *

    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 2c58b2ad..ca4cae75 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 @@ -2,6 +2,7 @@ package com.lingniu.ingest.protocol.gb32960.inbound; import com.lingniu.ingest.api.event.VehicleEvent; import com.lingniu.ingest.api.sink.EventSink; +import com.lingniu.ingest.codec.BccChecksum; import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor; import com.lingniu.ingest.core.concurrency.DisruptorEventBus; import com.lingniu.ingest.core.dispatcher.Dispatcher; @@ -21,6 +22,9 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.embedded.EmbeddedChannel; import org.junit.jupiter.api.Test; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -36,6 +40,7 @@ class Gb32960ChannelHandlerAckBoundaryTest { assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse(); assertThat((Object) channel.readOutbound()).isNull(); + dispatch.awaitPublishCount(1); dispatch.completeDurability(); channel.runPendingTasks(); @@ -55,6 +60,7 @@ class Gb32960ChannelHandlerAckBoundaryTest { assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse(); + dispatch.awaitPublishCount(1); dispatch.failDurability(); channel.runPendingTasks(); @@ -64,6 +70,35 @@ class Gb32960ChannelHandlerAckBoundaryTest { } } + @Test + void durableAcksAreWrittenInInboundOrderWhenLaterDispatchCompletesFirst() { + try (DispatchHarness dispatch = new DispatchHarness()) { + EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher())); + + assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse(); + assertThat(channel.writeInbound(buildHeartbeatFrame("LTEST000000000001"))).isFalse(); + assertThat((Object) channel.readOutbound()).isNull(); + + dispatch.awaitPublishCount(2); + dispatch.completeDurability(1); + channel.runPendingTasks(); + assertThat((Object) channel.readOutbound()).isNull(); + + dispatch.completeDurability(0); + channel.runPendingTasks(); + + ByteBuf firstAck = channel.readOutbound(); + ByteBuf secondAck = channel.readOutbound(); + assertThat(firstAck).isNotNull(); + assertThat(secondAck).isNotNull(); + assertThat(firstAck.getUnsignedByte(2)).isEqualTo((short) 0x02); + assertThat(secondAck.getUnsignedByte(2)).isEqualTo((short) 0x07); + firstAck.release(); + secondAck.release(); + channel.finishAndReleaseAll(); + } + } + private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) { return new Gb32960ChannelHandler( decoder(), @@ -87,6 +122,23 @@ class Gb32960ChannelHandlerAckBoundaryTest { new Gb32960PlatformAuthorizer(auth.getPlatforms())); } + private static byte[] buildHeartbeatFrame(String vin) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + out.write(0x23); + out.write(0x23); + out.write(0x07); + out.write(0xFE); + byte[] vinBytes = vin.getBytes(StandardCharsets.US_ASCII); + out.write(vinBytes, 0, 17); + out.write(0x01); + out.write(0x00); + out.write(0x00); + + byte[] almost = out.toByteArray(); + out.write(BccChecksum.compute(almost, 2, almost.length - 2) & 0xFF); + return out.toByteArray(); + } + private static final class DispatchHarness implements AutoCloseable { private final ControlledSink sink = new ControlledSink(); private final DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink)); @@ -103,11 +155,28 @@ class Gb32960ChannelHandlerAckBoundaryTest { } void completeDurability() { - sink.future.complete(null); + completeDurability(0); + } + + void completeDurability(int index) { + sink.future(index).complete(null); } void failDurability() { - sink.future.completeExceptionally(new IllegalStateException("durability failed")); + sink.future(0).completeExceptionally(new IllegalStateException("durability failed")); + } + + void awaitPublishCount(int count) { + long deadline = System.currentTimeMillis() + 3000; + while (System.currentTimeMillis() < deadline) { + synchronized (sink.futures) { + if (sink.futures.size() >= count) { + return; + } + } + Thread.onSpinWait(); + } + throw new AssertionError("expected " + count + " sink publishes, actual=" + sink.futures.size()); } @Override @@ -118,16 +187,26 @@ class Gb32960ChannelHandlerAckBoundaryTest { } private static final class ControlledSink implements EventSink { - private final CompletableFuture future = new CompletableFuture<>(); + private final List> futures = new ArrayList<>(); @Override public String name() { - return "controlled"; + return "kafka"; } @Override public CompletableFuture publish(VehicleEvent event) { + CompletableFuture future = new CompletableFuture<>(); + synchronized (futures) { + futures.add(future); + } return future; } + + private CompletableFuture future(int index) { + synchronized (futures) { + return futures.get(index); + } + } } }