fix: enforce gb32960 kafka ack boundary

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<VehicleEventSlot> disruptor;
     private final List<EventSink> sinks;
+    private final ExecutorService awaitExecutor;
     private final AtomicLong published = new AtomicLong();
     private final AtomicLong failed = new AtomicLong();
 
     public DisruptorEventBus(int ringBufferSize, String waitStrategyName, List<EventSink> 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<Void> publishAndAwait(VehicleEvent event) {
+    public CompletableFuture<Void> publishAndAwait(VehicleEvent event, String requiredSinkName) {
+        String requiredSink = Objects.requireNonNull(requiredSinkName, "requiredSinkName");
         published.incrementAndGet();
-        List<CompletableFuture<Void>> futures = sinks.stream()
-                .filter(sink -> sink.accepts(event))
-                .map(sink -> publishToSinkAndTrack(sink, event))
-                .toList();
-        if (futures.isEmpty()) {
-            return CompletableFuture.completedFuture(null);
+        List<CompletableFuture<Void>> requiredFutures = new ArrayList<>();
+        for (EventSink sink : sinks) {
+            if (!sink.accepts(event)) continue;
+
+            CompletableFuture<Void> 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());
     }
 

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<Void> 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<Void> 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<VehicleEvent> 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<VehicleEvent> events, Throwable failure) {
     }
+
+    private static final class AwaitedAsyncBatchUnsupportedException extends IllegalStateException {
+        private AwaitedAsyncBatchUnsupportedException(String message) {
+            super(message);
+        }
+    }
 }