From 6a8b53bab65852107734b6e51bf826009f52b3b5 Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 23 Jun 2026 17:43:28 +0800 Subject: [PATCH] feat: split history and analytics consumers --- .../TelemetryEnvelopeRecordMapper.java | 35 +++++++++++++++-- .../EventHistoryEnvelopeIngestorTest.java | 39 +++++++++++++++---- .../VehicleStatEnvelopeIngestor.java | 6 ++- .../VehicleStatEnvelopeIngestorTest.java | 32 +++++++++++++++ .../VehicleStateEnvelopeIngestor.java | 10 ++++- .../VehicleStateEnvelopeIngestorTest.java | 34 ++++++++++++++++ 6 files changed, 143 insertions(+), 13 deletions(-) diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java index 4f6fe5b3..905d0842 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java @@ -60,8 +60,14 @@ public final class TelemetryEnvelopeRecordMapper { ProtocolId protocol = protocol(envelope.getSource()); RawArchiveRef rawArchive = envelope.getRawArchive(); Map metadata = metadata(envelope, protocol); - String rawArchiveUri = rawArchive.getUri(); + String rawArchiveKey = rawArchiveKey(rawArchive.getUri(), metadata); + String rawArchiveUri = rawArchive.getUri().isBlank() && !rawArchiveKey.isBlank() + ? RawArchiveKeys.logicalUri(rawArchiveKey) + : rawArchive.getUri(); metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, envelope.getEventId()); + if (!rawArchiveKey.isBlank()) { + metadata.putIfAbsent(RawArchiveKeys.META_KEY, rawArchiveKey); + } if (!rawArchiveUri.isBlank()) { metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri); } @@ -74,10 +80,21 @@ public final class TelemetryEnvelopeRecordMapper { Instant.ofEpochMilli(envelope.getIngestTimeMs()), rawArchiveUri, metadata, - rawArchiveJson(envelope, metadata) + rawArchiveJson(envelope, metadata, rawArchiveKey, rawArchiveUri) ); } + private static String rawArchiveKey(String rawArchiveUri, Map metadata) { + String key = metadata.getOrDefault(RawArchiveKeys.META_KEY, ""); + if (!key.isBlank()) { + return key; + } + if (rawArchiveUri != null && rawArchiveUri.startsWith("archive://")) { + return rawArchiveUri.substring("archive://".length()); + } + return ""; + } + private static Map metadata(VehicleEnvelope envelope, ProtocolId protocol) { Map metadata = new LinkedHashMap<>(envelope.getMetadataMap()); if (!envelope.getProtocolVersion().isBlank()) { @@ -109,7 +126,10 @@ public final class TelemetryEnvelopeRecordMapper { } } - private static String rawArchiveJson(VehicleEnvelope envelope, Map metadata) { + private static String rawArchiveJson(VehicleEnvelope envelope, + Map metadata, + String rawArchiveKey, + String rawArchiveUri) { Map payload = new LinkedHashMap<>(); RawArchiveRef rawArchive = envelope.getRawArchive(); payload.put("eventId", envelope.getEventId()); @@ -121,7 +141,14 @@ public final class TelemetryEnvelopeRecordMapper { payload.put("eventType", "RAW_ARCHIVE"); payload.put("eventTime", Instant.ofEpochMilli(envelope.getEventTimeMs()).toString()); payload.put("ingestTime", Instant.ofEpochMilli(envelope.getIngestTimeMs()).toString()); - payload.put("rawArchiveUri", rawArchive.getUri()); + payload.put("rawArchiveUri", rawArchiveUri); + payload.put("rawArchiveKey", rawArchiveKey); + if (metadata.containsKey("command")) { + payload.put("command", metadata.get("command")); + } + if (metadata.containsKey("infoType")) { + payload.put("infoType", metadata.get("infoType")); + } payload.put("rawSizeBytes", rawArchive.getSizeBytes()); payload.put("metadata", metadata); try { diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java index b391f9dd..6a142bda 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java @@ -1,7 +1,11 @@ package com.lingniu.ingest.eventhistory; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.lingniu.ingest.api.ProtocolId; import com.lingniu.ingest.api.consumer.EnvelopeIngestResult; import com.lingniu.ingest.api.consumer.EnvelopeIngestor; +import com.lingniu.ingest.api.event.RawArchiveKeys; import com.lingniu.ingest.eventfilestore.EventFileQuery; import com.lingniu.ingest.eventfilestore.EventFileRecord; import com.lingniu.ingest.eventfilestore.EventFileStore; @@ -12,6 +16,7 @@ import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; import org.junit.jupiter.api.Test; import java.io.IOException; +import java.time.Instant; import java.util.ArrayList; import java.util.List; @@ -20,6 +25,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; class EventHistoryEnvelopeIngestorTest { + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + @Test void parsesKafkaEnvelopeBytesAndAppendsRecord() throws Exception { CapturingStore store = new CapturingStore(); @@ -57,10 +64,12 @@ class EventHistoryEnvelopeIngestorTest { } @Test - void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() { + void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() throws Exception { CapturingStore store = new CapturingStore(); EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper()); + String rawArchiveKey = "2026/06/23/GB32960/VINRAW001/raw-event-1.bin"; + String rawArchiveUri = "archive://" + rawArchiveKey; VehicleEnvelope envelope = VehicleEnvelope.newBuilder() .setSchemaVersion("1.0") @@ -70,9 +79,11 @@ class EventHistoryEnvelopeIngestorTest { .setProtocolVersion("V2016") .setEventTimeMs(1_782_112_400_000L) .setIngestTimeMs(1_782_112_401_000L) - .putMetadata("archiveKey", "gb32960/2026/06/23/raw-event-1.bin") + .putMetadata(RawArchiveKeys.META_KEY, rawArchiveKey) + .putMetadata(RawArchiveKeys.META_URI, rawArchiveUri) + .putMetadata("command", "0x0002") .setRawArchive(RawArchiveRef.newBuilder() - .setUri("archive://gb32960/2026/06/23/raw-event-1.bin") + .setUri(rawArchiveUri) .setSizeBytes(128) .build()) .build(); @@ -83,14 +94,28 @@ class EventHistoryEnvelopeIngestorTest { assertThat(store.records).hasSize(1); EventFileRecord record = store.records.getFirst(); assertThat(record.eventId()).isEqualTo("raw-event-1"); + assertThat(record.protocol()).isEqualTo(ProtocolId.GB32960); assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE"); assertThat(record.vin()).isEqualTo("VINRAW001"); - assertThat(record.rawArchiveUri()).isEqualTo("archive://gb32960/2026/06/23/raw-event-1.bin"); + assertThat(record.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_112_400_000L)); + assertThat(record.rawArchiveUri()).isEqualTo(rawArchiveUri); assertThat(record.metadata()) .containsEntry("protocolVersion", "V2016") - .containsEntry("archiveKey", "gb32960/2026/06/23/raw-event-1.bin"); - assertThat(record.payloadJson()).contains("\"rawSizeBytes\":128"); - assertThat(record.payloadJson()).doesNotContain("rawBytes"); + .containsEntry(RawArchiveKeys.META_EVENT_ID, "raw-event-1") + .containsEntry(RawArchiveKeys.META_KEY, rawArchiveKey) + .containsEntry(RawArchiveKeys.META_URI, rawArchiveUri) + .containsEntry("command", "0x0002"); + JsonNode payload = OBJECT_MAPPER.readTree(record.payloadJson()); + assertThat(payload.get("eventId").asText()).isEqualTo("raw-event-1"); + assertThat(payload.get("protocol").asText()).isEqualTo("GB32960"); + assertThat(payload.get("eventType").asText()).isEqualTo("RAW_ARCHIVE"); + assertThat(payload.get("vin").asText()).isEqualTo("VINRAW001"); + assertThat(payload.get("eventTime").asText()).isEqualTo("2026-06-22T07:13:20Z"); + assertThat(payload.get("rawArchiveUri").asText()).isEqualTo(rawArchiveUri); + assertThat(payload.get("rawArchiveKey").asText()).isEqualTo(rawArchiveKey); + assertThat(payload.get("rawSizeBytes").asLong()).isEqualTo(128); + assertThat(payload.get("metadata").get("command").asText()).isEqualTo("0x0002"); + assertThat(payload.toString()).doesNotContain("rawBytes"); } private static VehicleEnvelope envelope(String eventId) { diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java index 19e36e34..aa8d2b39 100644 --- a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java @@ -17,7 +17,11 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor { } public void ingest(byte[] kafkaValue) { - processor.process(parse(kafkaValue)); + VehicleEnvelope envelope = parse(kafkaValue); + if (!envelope.hasTelemetrySnapshot()) { + return; + } + processor.process(envelope); } @Override diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java index 3f80f9b5..efceed2e 100644 --- a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java @@ -2,6 +2,7 @@ package com.lingniu.ingest.vehiclestat; import com.lingniu.ingest.api.consumer.EnvelopeIngestResult; import com.lingniu.ingest.api.consumer.EnvelopeIngestor; +import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; import com.lingniu.ingest.sink.mq.proto.TelemetryField; import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot; import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; @@ -74,6 +75,23 @@ class VehicleStatEnvelopeIngestorTest { assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty(); } + @Test + void rawArchiveEnvelopeIsIgnoredWithoutAppendingMileagePoint() { + InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository(); + VehicleStatEnvelopeIngestor ingestor = + new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository)); + VehicleEnvelope envelope = rawArchiveEnvelope(); + + ingestor.ingest(envelope.toByteArray()); + var result = ingestor.tryIngest(envelope.toByteArray()); + + assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED); + assertThat(result.eventId()).isEqualTo("raw-event-1"); + assertThat(result.vin()).isEqualTo("VIN001"); + assertThat(result.message()).contains("telemetry_snapshot"); + assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty(); + } + private static VehicleEnvelope envelope() { return VehicleEnvelope.newBuilder() .setEventId("event-1") @@ -91,6 +109,20 @@ class VehicleStatEnvelopeIngestorTest { .build(); } + private static VehicleEnvelope rawArchiveEnvelope() { + return VehicleEnvelope.newBuilder() + .setEventId("raw-event-1") + .setVin("VIN001") + .setSource("GB32960") + .setEventTimeMs(1_782_112_400_000L) + .setIngestTimeMs(1_782_112_401_000L) + .setRawArchive(RawArchiveRef.newBuilder() + .setUri("archive://2026/06/23/GB32960/VIN001/raw-event-1.bin") + .setSizeBytes(128) + .build()) + .build(); + } + private static final class InMemoryVehicleStatRepository implements VehicleStatRepository { private final java.util.Map> points = new java.util.HashMap<>(); diff --git a/modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java b/modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java index 8251ce3d..f4e4489e 100644 --- a/modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java +++ b/modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java @@ -17,7 +17,11 @@ public final class VehicleStateEnvelopeIngestor implements EnvelopeIngestor { } public void ingest(byte[] kafkaValue) { - updater.update(parse(kafkaValue)); + VehicleEnvelope envelope = parse(kafkaValue); + if (!envelope.hasTelemetrySnapshot()) { + return; + } + updater.update(envelope); } @Override @@ -25,6 +29,10 @@ public final class VehicleStateEnvelopeIngestor implements EnvelopeIngestor { VehicleEnvelope envelope = null; try { envelope = parse(kafkaValue); + if (!envelope.hasTelemetrySnapshot()) { + return EnvelopeIngestResult.skipped( + envelope.getEventId(), envelope.getVin(), "envelope telemetry_snapshot is required"); + } // Kafka 消费路径只接受标准 VehicleEnvelope；坏消息返回明确结果给处理器写 DLQ。 updater.update(envelope); return EnvelopeIngestResult.processed(envelope.getEventId(), envelope.getVin()); diff --git a/modules/services/vehicle-state-service/src/test/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestorTest.java b/modules/services/vehicle-state-service/src/test/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestorTest.java index 8a8dfcc1..6f5ec96b 100644 --- a/modules/services/vehicle-state-service/src/test/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestorTest.java +++ b/modules/services/vehicle-state-service/src/test/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestorTest.java @@ -2,6 +2,7 @@ package com.lingniu.ingest.vehiclestate; import com.lingniu.ingest.api.consumer.EnvelopeIngestResult; import com.lingniu.ingest.api.consumer.EnvelopeIngestor; +import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; import com.lingniu.ingest.sink.mq.proto.TelemetryField; import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot; import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; @@ -47,6 +48,25 @@ class VehicleStateEnvelopeIngestorTest { assertThat(repository.getState("VIN001")).isEmpty(); } + @Test + void rawArchiveEnvelopeIsIgnoredWithoutUpdatingRepository() { + InMemoryVehicleStateRepository repository = new InMemoryVehicleStateRepository(); + VehicleStateEnvelopeIngestor ingestor = new VehicleStateEnvelopeIngestor(new VehicleStateUpdater(repository)); + VehicleEnvelope envelope = rawArchiveEnvelope(); + + ingestor.ingest(envelope.toByteArray()); + var result = ingestor.tryIngest(envelope.toByteArray()); + + assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED); + assertThat(result.eventId()).isEqualTo("raw-event-1"); + assertThat(result.vin()).isEqualTo("VIN001"); + assertThat(result.message()).contains("telemetry_snapshot"); + assertThat(repository.getState("VIN001")).isEmpty(); + assertThat(repository.getLocation("VIN001")).isEmpty(); + assertThat(repository.getSafety("VIN001")).isEmpty(); + assertThat(repository.getLastEvent("VIN001")).isEmpty(); + } + private static VehicleEnvelope envelope() { return VehicleEnvelope.newBuilder() .setEventId("event-1") @@ -64,6 +84,20 @@ class VehicleStateEnvelopeIngestorTest { .build(); } + private static VehicleEnvelope rawArchiveEnvelope() { + return VehicleEnvelope.newBuilder() + .setEventId("raw-event-1") + .setVin("VIN001") + .setSource("GB32960") + .setEventTimeMs(1_782_112_400_000L) + .setIngestTimeMs(1_782_112_401_000L) + .setRawArchive(RawArchiveRef.newBuilder() + .setUri("archive://2026/06/23/GB32960/VIN001/raw-event-1.bin") + .setSizeBytes(128) + .build()) + .build(); + } + private static final class EmptyRepository implements VehicleStateRepository { @Override public void putState(String vin, String json) {} @Override public void putLocation(String vin, String json) {}