diff --git a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java index e58c798d..bf590dde 100644 --- a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java +++ b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java @@ -43,13 +43,11 @@ public final class KafkaEventSink implements EventSink, AutoCloseable { } /** - * 拒绝 {@link VehicleEvent.RawArchive} —— 当前生产 Kafka sink 只投递实时、会话、告警等轻量事件。 - * {@link EnvelopeMapper} 和 {@link TopicRouter} 已具备 RAW envelope/topic 的映射能力，但这里仍 - * 显式过滤；如需把 32960 raw 引用转发到 {@code vehicle.raw.archive}，应先移除此过滤并补充测试。 + * Split-service mode uses Kafka for both normalized events and raw archive records. */ @Override public boolean accepts(VehicleEvent event) { - return !(event instanceof VehicleEvent.RawArchive); + return true; } @Override diff --git a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java index 632be968..f4c46171 100644 --- a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java +++ b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java @@ -57,16 +57,16 @@ public class SinkMqProperties { public static class Topics { /** 实时遥测事件 topic；GB32960 RAW-only 架构下可逐步弱化该 topic。 */ - private String realtime = "vehicle.realtime"; + private String realtime = "vehicle.event.gb32960.v1"; /** 位置事件 topic；通常可由 realtime/RAW 派生。 */ - private String location = "vehicle.location"; - private String alarm = "vehicle.alarm"; - private String session = "vehicle.session"; - private String mediaMeta = "vehicle.media.meta"; + private String location = "vehicle.event.gb32960.v1"; + private String alarm = "vehicle.event.gb32960.v1"; + private String session = "vehicle.event.gb32960.v1"; + private String mediaMeta = "vehicle.media.meta.v1"; /** RAW 归档引用 topic，payload 应携带 archive URI/size，不建议携带完整 raw bytes。 */ - private String rawArchive = "vehicle.raw.archive"; + private String rawArchive = "vehicle.raw.gb32960.v1"; /** producer 熔断或 consumer 处理失败时的死信 topic。 */ - private String dlq = "vehicle.dlq"; + private String dlq = "vehicle.dlq.gb32960.v1"; public String getRealtime() { return realtime; } public void setRealtime(String realtime) { this.realtime = realtime; } diff --git a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java new file mode 100644 index 00000000..c85154a2 --- /dev/null +++ b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java @@ -0,0 +1,40 @@ +package com.lingniu.ingest.sink.mq; + +import com.lingniu.ingest.api.ProtocolId; +import com.lingniu.ingest.api.event.VehicleEvent; +import io.github.resilience4j.circuitbreaker.CircuitBreaker; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class KafkaEventSinkTest { + + @Test + void acceptsRawArchiveRecordsForSplitServiceKafkaBoundary() { + KafkaEventSink sink = new KafkaEventSink( + null, + new EnvelopeMapper("node-1"), + new TopicRouter(new SinkMqProperties.Topics()), + "vehicle.dlq.gb32960.v1", + CircuitBreaker.ofDefaults("kafka-test")); + + assertThat(sink.accepts(rawArchive())).isTrue(); + } + + private static VehicleEvent.RawArchive rawArchive() { + return new VehicleEvent.RawArchive( + "raw-1", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-raw-1", + Map.of("platformAccount", "Hyundai"), + 0x02, + 0, + new byte[] {0x23, 0x23}); + } +} diff --git a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java new file mode 100644 index 00000000..7ba6c418 --- /dev/null +++ b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java @@ -0,0 +1,147 @@ +package com.lingniu.ingest.sink.mq; + +import com.lingniu.ingest.api.ProtocolId; +import com.lingniu.ingest.api.event.AlarmPayload; +import com.lingniu.ingest.api.event.LocationPayload; +import com.lingniu.ingest.api.event.RealtimePayload; +import com.lingniu.ingest.api.event.VehicleEvent; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; + +class TopicRouterTest { + + @Test + void rawArchiveRoutesToVersionedGb32960RawTopicByDefault() { + TopicRouter router = new TopicRouter(new SinkMqProperties.Topics()); + + assertThat(router.route(rawArchive())).isEqualTo("vehicle.raw.gb32960.v1"); + } + + @Test + void normalizedGb32960EventsRouteToVersionedEventTopicByDefault() { + TopicRouter router = new TopicRouter(new SinkMqProperties.Topics()); + + assertThat(router.route(realtime())).isEqualTo("vehicle.event.gb32960.v1"); + assertThat(router.route(location())).isEqualTo("vehicle.event.gb32960.v1"); + assertThat(router.route(alarm())).isEqualTo("vehicle.event.gb32960.v1"); + assertThat(router.route(login())).isEqualTo("vehicle.event.gb32960.v1"); + assertThat(router.route(logout())).isEqualTo("vehicle.event.gb32960.v1"); + assertThat(router.route(heartbeat())).isEqualTo("vehicle.event.gb32960.v1"); + } + + @Test + void dlqDefaultsToVersionedGb32960DlqTopic() { + SinkMqProperties.Topics topics = new SinkMqProperties.Topics(); + + assertThat(topics.getDlq()).isEqualTo("vehicle.dlq.gb32960.v1"); + } + + private static VehicleEvent.RawArchive rawArchive() { + return new VehicleEvent.RawArchive( + "raw-1", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-raw-1", + Map.of("platformAccount", "Hyundai"), + 0x02, + 0, + new byte[] {0x23, 0x23}); + } + + private static VehicleEvent.Realtime realtime() { + return new VehicleEvent.Realtime( + "event-1", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-1", + Map.of("platformAccount", "Hyundai"), + new RealtimePayload( + 42.1, 1234.5, 80.0, null, null, + null, null, null, null, null, null, + RealtimePayload.VehicleState.STARTED, + RealtimePayload.ChargingState.UNCHARGED, + RealtimePayload.RunningMode.ELECTRIC, + null, null, null, + 113.12, 23.45, null, null, null, null)); + } + + private static VehicleEvent.Location location() { + return new VehicleEvent.Location( + "event-2", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-2", + Map.of("platformAccount", "Hyundai"), + new LocationPayload(113.12, 23.45, 0, 42.1, 90, 0, 0)); + } + + private static VehicleEvent.Alarm alarm() { + return new VehicleEvent.Alarm( + "event-3", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-3", + Map.of("platformAccount", "Hyundai"), + new AlarmPayload( + AlarmPayload.AlarmLevel.MINOR, + 1, + "GB32960_ALARM", + List.of("1"), + Set.of("bit0"), + 113.12, + 23.45, + AlarmPayload.SafetyCategory.GENERAL, + false, + AlarmPayload.HydrogenLeakLevel.NONE, + false)); + } + + private static VehicleEvent.Login login() { + return new VehicleEvent.Login( + "event-4", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-4", + Map.of("platformAccount", "Hyundai"), + "iccid-1", + "V2016"); + } + + private static VehicleEvent.Logout logout() { + return new VehicleEvent.Logout( + "event-5", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-5", + Map.of("platformAccount", "Hyundai")); + } + + private static VehicleEvent.Heartbeat heartbeat() { + return new VehicleEvent.Heartbeat( + "event-6", + "VIN001", + ProtocolId.GB32960, + Instant.parse("2026-06-23T07:00:00Z"), + Instant.parse("2026-06-23T07:00:01Z"), + "trace-6", + Map.of("platformAccount", "Hyundai")); + } +}