shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: prevent tdengine raw frame overwrite
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 18:41:58 +08:00
parent 5c1c61f29b
commit 8804c01d99
9 changed files with 422 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java
View File

@@ -30,7 +30,7 @@ public final class TdengineEnvelopeRows {
metadata.putAll(envelope.getMetadataMap());
metadata.putAll(raw.getMetadataMap());
return Optional.of(new TdengineRawFrameRow(
instant(envelope.getEventTimeMs()),
rawFrameTs(envelope, metadata),
raw.getFrameId(),
instant(envelope.getIngestTimeMs()),
raw.getMessageId(),
@@ -142,6 +142,24 @@ public final class TdengineEnvelopeRows {
return Instant.ofEpochMilli(epochMillis);
}
private static Instant rawFrameTs(VehicleEnvelope envelope, Map<String, String> metadata) {
String rawArchiveEventId = firstNonBlank(metadata.get("rawArchiveEventId"), envelope.getEventId());
if (rawArchiveEventId == null || rawArchiveEventId.isBlank()) {
return instant(envelope.getEventTimeMs());
}
try {
long sequenceId = Long.parseLong(rawArchiveEventId);
long baseMillis = sequenceId / 1000L;
long sameMillisecondOffset = sequenceId % 1000L;
if (Math.abs(baseMillis - envelope.getEventTimeMs()) > 60_000L) {
return instant(envelope.getEventTimeMs());
}
return Instant.ofEpochMilli(baseMillis + sameMillisecondOffset);
} catch (NumberFormatException ignored) {
return instant(envelope.getEventTimeMs());
}
}
private static String protocol(VehicleEnvelope envelope) {
return firstNonBlank(envelope.getSource(), "UNKNOWN");
}

35
modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java
View File

@@ -51,6 +51,20 @@ class TdengineEnvelopeRowsTest {
assertThat(row.metadataJson()).contains("\"channel\":\"tcp-808\"", "\"auth\":\"passed\"");
}
@Test
void expandsRawFrameTsWithRawArchiveEventIdSequenceToAvoidSameMillisecondOverwrite() {
VehicleEnvelope first = rawFrameEnvelope("1782729152494000");
VehicleEnvelope second = rawFrameEnvelope("1782729152494001");
TdengineRawFrameRow firstRow = TdengineEnvelopeRows.rawFrame(first).orElseThrow();
TdengineRawFrameRow secondRow = TdengineEnvelopeRows.rawFrame(second).orElseThrow();
assertThat(firstRow.ts()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_494L));
assertThat(secondRow.ts()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_495L));
assertThat(firstRow.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_494L));
assertThat(secondRow.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_494L));
}
@Test
void mapsLocationEnvelopeToVehicleLocationRow() {
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
@@ -190,4 +204,25 @@ class TdengineEnvelopeRowsTest {
assertThat(TdengineEnvelopeRows.telemetryFields(envelope).getFirst().vehicleKey())
.isEqualTo("jt808:13079963291");
}
private static VehicleEnvelope rawFrameEnvelope(String rawArchiveEventId) {
return VehicleEnvelope.newBuilder()
.setEventId(rawArchiveEventId)
.setVin("unknown")
.setSource("JT808")
.setEventTimeMs(1_782_729_152_494L)
.setIngestTimeMs(1_782_729_152_539L)
.putMetadata("phone", "13079973200")
.putMetadata("rawArchiveEventId", rawArchiveEventId)
.setRawFrameFact(RawFrameFactPayload.newBuilder()
.setFrameId("rf-" + rawArchiveEventId)
.setPhone("13079973200")
.setMessageId(0x0200)
.setRawUri("archive://2026/06/29/JT808/unknown/" + rawArchiveEventId + ".bin")
.setChecksum("sha256:" + rawArchiveEventId)
.setRawSizeBytes(41)
.setParseStatus(ParseStatusProto.PARSE_STATUS_NOT_PARSED)
.putMetadata("rawArchiveEventId", rawArchiveEventId))
.build();
}
}