diff --git a/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java b/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java index 69fd7f1d..4d01b568 100644 --- a/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java +++ b/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java @@ -11,6 +11,7 @@ import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerWorker; import com.lingniu.ingest.sink.mq.SinkMqProperties; import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -32,10 +33,17 @@ public class VehicleHistoryKafkaConsumerConfiguration { @Bean @ConditionalOnMissingBean - public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(EventFileStore store, + public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(ObjectProvider store, TelemetryEnvelopeRecordMapper mapper, - ObjectProvider writer) { - return new EventHistoryEnvelopeIngestor(store, mapper, writer.getIfAvailable()); + ObjectProvider writer, + @Value("${lingniu.ingest.tdengine-history.telemetry-fields-enabled:false}") + boolean telemetryFieldsEnabled) { + TdengineHistoryWriter tdengineWriter = writer.getIfAvailable(); + EventFileStore eventFileStore = store.getIfAvailable(); + if (eventFileStore == null) { + return new EventHistoryEnvelopeIngestor(mapper, tdengineWriter, telemetryFieldsEnabled); + } + return new EventHistoryEnvelopeIngestor(eventFileStore, mapper, tdengineWriter, telemetryFieldsEnabled); } @Bean @@ -62,7 +70,11 @@ public class VehicleHistoryKafkaConsumerConfiguration { List workers = new KafkaEnvelopeConsumerFactory().createWorkers( Map.of( "eventHistoryEnvelopeConsumerProcessor", processor, - "eventHistoryRawEnvelopeConsumerProcessor", rawProcessor), + "eventHistoryGb32960EnvelopeConsumerProcessor", processor, + "eventHistoryJt808EnvelopeConsumerProcessor", processor, + "eventHistoryRawEnvelopeConsumerProcessor", rawProcessor, + "eventHistoryGb32960RawEnvelopeConsumerProcessor", rawProcessor, + "eventHistoryJt808RawEnvelopeConsumerProcessor", rawProcessor), props); if (workers.isEmpty()) { throw new IllegalStateException("no vehicle history kafka consumer workers created; check consumer bindings"); diff --git a/modules/apps/vehicle-history-app/src/main/resources/application.yml b/modules/apps/vehicle-history-app/src/main/resources/application.yml index 867fc42e..270ac15a 100644 --- a/modules/apps/vehicle-history-app/src/main/resources/application.yml +++ b/modules/apps/vehicle-history-app/src/main/resources/application.yml @@ -48,25 +48,34 @@ lingniu: auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest} max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:100} max-poll-interval-millis: ${KAFKA_CONSUMER_MAX_POLL_INTERVAL_MS:1800000} + concurrency: ${KAFKA_CONSUMER_CONCURRENCY:3} bindings: - eventHistoryEnvelopeConsumerProcessor: + eventHistoryGb32960EnvelopeConsumerProcessor: enabled: true - group-id: ${KAFKA_GROUP_HISTORY_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-event} + group-id: ${KAFKA_GROUP_HISTORY_GB32960_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-gb32960-event} topics: - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} - - ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} - eventHistoryRawEnvelopeConsumerProcessor: + eventHistoryJt808EnvelopeConsumerProcessor: enabled: true - group-id: ${KAFKA_GROUP_HISTORY_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}} + group-id: ${KAFKA_GROUP_HISTORY_JT808_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-event} + topics: + - ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} + eventHistoryGb32960RawEnvelopeConsumerProcessor: + enabled: true + group-id: ${KAFKA_GROUP_HISTORY_GB32960_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-gb32960-raw} topics: - ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1} + eventHistoryJt808RawEnvelopeConsumerProcessor: + enabled: true + group-id: ${KAFKA_GROUP_HISTORY_JT808_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-raw} + topics: - ${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1} archive: enabled: ${SINK_ARCHIVE_ENABLED:true} type: local path: ${SINK_ARCHIVE_PATH:./archive/} event-file-store: - enabled: ${EVENT_FILE_STORE_ENABLED:true} + enabled: ${EVENT_FILE_STORE_ENABLED:false} path: ${EVENT_FILE_STORE_PATH:./target/event-store/} zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai} batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:500} @@ -82,6 +91,7 @@ lingniu: minimum-idle: ${TDENGINE_MIN_IDLE:0} connection-timeout-millis: ${TDENGINE_CONNECTION_TIMEOUT_MILLIS:5000} initialization-fail-timeout-millis: ${TDENGINE_INITIALIZATION_FAIL_TIMEOUT_MILLIS:-1} + telemetry-fields-enabled: ${TDENGINE_TELEMETRY_FIELDS_ENABLED:false} event-history: enabled: true vehicle-state: diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java index db4034d8..71d7ffca 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java @@ -3,6 +3,7 @@ package com.lingniu.ingest.historyapp; import com.lingniu.ingest.eventfilestore.EventFileStore; import com.lingniu.ingest.eventfilestore.EventFileStoreSink; import com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration; +import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink; import com.lingniu.ingest.eventhistory.EventHistoryController; import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; @@ -20,6 +21,7 @@ import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink; import com.lingniu.ingest.sink.mq.KafkaEventSink; import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration; import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema; +import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter; import com.lingniu.ingest.tdenginehistory.config.TdengineHistoryAutoConfiguration; import org.apache.kafka.clients.producer.KafkaProducer; import org.junit.jupiter.api.Test; @@ -86,6 +88,21 @@ class VehicleHistoryAppCompositionTest { }); } + @Test + void createsHistoryIngestorWithoutEventFileStoreForTdengineOnlyRuntime() { + new ApplicationContextRunner() + .withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class) + .withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class)) + .withBean(EnvelopeDeadLetterSink.class, () -> mock(EnvelopeDeadLetterSink.class)) + .withPropertyValues( + "lingniu.ingest.event-file-store.enabled=false", + "lingniu.ingest.sink.mq.consumer.enabled=false") + .run(context -> { + assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class); + assertThat(context).doesNotHaveBean(EventFileStore.class); + }); + } + @SuppressWarnings("unchecked") private static KafkaProducer kafkaProducer() { return mock(KafkaProducer.class); diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java index 0b2b4b01..cee8c987 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java @@ -23,7 +23,7 @@ class VehicleHistoryAppDefaultsTest { .containsEntry("lingniu.ingest.gb32960.enabled", true) .containsEntry("lingniu.ingest.gb32960.server.enabled", false) .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}") - .containsEntry("lingniu.ingest.event-file-store.enabled", "${EVENT_FILE_STORE_ENABLED:true}") + .containsEntry("lingniu.ingest.event-file-store.enabled", "${EVENT_FILE_STORE_ENABLED:false}") .containsEntry("lingniu.ingest.tdengine-history.enabled", "${TDENGINE_HISTORY_ENABLED:false}") .containsEntry("lingniu.ingest.tdengine-history.database", "${TDENGINE_HISTORY_DATABASE:vehicle_history}") .containsEntry( @@ -36,37 +36,53 @@ class VehicleHistoryAppDefaultsTest { "${TDENGINE_DRIVER_CLASS_NAME:com.taosdata.jdbc.rs.RestfulDriver}") .containsEntry("lingniu.ingest.tdengine-history.maximum-pool-size", "${TDENGINE_MAX_POOL_SIZE:16}") .containsEntry("lingniu.ingest.tdengine-history.minimum-idle", "${TDENGINE_MIN_IDLE:0}") + .containsEntry( + "lingniu.ingest.tdengine-history.telemetry-fields-enabled", + "${TDENGINE_TELEMETRY_FIELDS_ENABLED:false}") .containsEntry("lingniu.ingest.event-history.enabled", true) .containsEntry("lingniu.ingest.vehicle-state.enabled", false) .containsEntry("lingniu.ingest.vehicle-stat.enabled", false) .containsEntry("lingniu.ingest.sink.mq.consumer.enabled", "${KAFKA_CONSUMER_ENABLED:true}") .containsEntry("lingniu.ingest.sink.mq.consumer.max-poll-records", "${KAFKA_CONSUMER_MAX_POLL_RECORDS:100}") + .containsEntry("lingniu.ingest.sink.mq.consumer.concurrency", "${KAFKA_CONSUMER_CONCURRENCY:3}") .containsEntry( "lingniu.ingest.sink.mq.consumer.max-poll-interval-millis", "${KAFKA_CONSUMER_MAX_POLL_INTERVAL_MS:1800000}") .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.enabled", + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960EnvelopeConsumerProcessor.enabled", true) .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.group-id", - "${KAFKA_GROUP_HISTORY_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-event}") + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960EnvelopeConsumerProcessor.group-id", + "${KAFKA_GROUP_HISTORY_GB32960_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-gb32960-event}") .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.topics[0]", + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960EnvelopeConsumerProcessor.topics[0]", "${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}") .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.topics[1]", - "${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}") - .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryRawEnvelopeConsumerProcessor.enabled", + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808EnvelopeConsumerProcessor.enabled", true) .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryRawEnvelopeConsumerProcessor.group-id", - "${KAFKA_GROUP_HISTORY_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}}") + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808EnvelopeConsumerProcessor.group-id", + "${KAFKA_GROUP_HISTORY_JT808_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-event}") .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryRawEnvelopeConsumerProcessor.topics[0]", + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808EnvelopeConsumerProcessor.topics[0]", + "${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}") + .containsEntry( + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960RawEnvelopeConsumerProcessor.enabled", + true) + .containsEntry( + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960RawEnvelopeConsumerProcessor.group-id", + "${KAFKA_GROUP_HISTORY_GB32960_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-gb32960-raw}") + .containsEntry( + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960RawEnvelopeConsumerProcessor.topics[0]", "${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}") .containsEntry( - "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryRawEnvelopeConsumerProcessor.topics[1]", + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808RawEnvelopeConsumerProcessor.enabled", + true) + .containsEntry( + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808RawEnvelopeConsumerProcessor.group-id", + "${KAFKA_GROUP_HISTORY_JT808_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-raw}") + .containsEntry( + "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808RawEnvelopeConsumerProcessor.topics[0]", "${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}"); } diff --git a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptor.java b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptor.java index 12efe9b4..3a7da48d 100644 --- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptor.java +++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptor.java @@ -36,16 +36,44 @@ public class RateLimitInterceptor implements IngestInterceptor, Ordered { @Override public boolean before(RawFrame frame, IngestContext ctx) { - String vin = frame.sourceMeta().getOrDefault("vin", "unknown"); - // VIN 粒度隔离，避免某一辆车的高频/异常上报挤占其他车辆处理额度。 - RateLimiter rl = limiters.get(vin, k -> RateLimiter.of("vin-" + k, defaultConfig)); + String limiterKey = limiterKey(frame); + // 车辆粒度隔离：VIN 可用时按 VIN；JT808 等未映射 VIN 的协议按 vehicleKey/phone 兜底， + // 避免所有 unknown 终端挤在同一个限流桶里。 + RateLimiter rl = limiters.get(limiterKey, k -> RateLimiter.of("vehicle-" + k, defaultConfig)); if (!rl.acquirePermission()) { - ctx.abort("rate-limited:" + vin); + ctx.abort("rate-limited:" + limiterKey); return false; } return true; } + private static String limiterKey(RawFrame frame) { + String vin = meta(frame, "vin"); + if (isKnown(vin)) { + return "vin:" + vin; + } + String vehicleKey = meta(frame, "vehicleKey"); + if (isKnown(vehicleKey)) { + return "vehicleKey:" + vehicleKey; + } + String phone = meta(frame, "phone"); + if (isKnown(phone)) { + return "phone:" + phone; + } + return "unknown"; + } + + private static String meta(RawFrame frame, String key) { + if (frame == null || frame.sourceMeta() == null) { + return ""; + } + return frame.sourceMeta().getOrDefault(key, "").trim(); + } + + private static boolean isKnown(String value) { + return value != null && !value.isBlank() && !"unknown".equalsIgnoreCase(value.trim()); + } + @Override public int getOrder() { return 200; diff --git a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptorTest.java b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptorTest.java new file mode 100644 index 00000000..45859a3e --- /dev/null +++ b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptorTest.java @@ -0,0 +1,55 @@ +package com.lingniu.ingest.core.pipeline.builtin; + +import com.lingniu.ingest.api.ProtocolId; +import com.lingniu.ingest.api.pipeline.IngestContext; +import com.lingniu.ingest.api.pipeline.RawFrame; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class RateLimitInterceptorTest { + + @Test + void unknownJt808VinUsesPhoneAsIndependentLimiterKey() { + RateLimitInterceptor interceptor = new RateLimitInterceptor(1, 100); + IngestContext first = new IngestContext("trace-1"); + IngestContext second = new IngestContext("trace-2"); + + boolean firstAllowed = interceptor.before(frame("unknown", "13079970001"), first); + boolean secondAllowed = interceptor.before(frame("unknown", "13079970002"), second); + + assertThat(firstAllowed).isTrue(); + assertThat(secondAllowed).isTrue(); + assertThat(first.aborted()).isFalse(); + assertThat(second.aborted()).isFalse(); + } + + @Test + void sameJt808PhoneStillSharesLimiterWhenVinIsUnknown() { + RateLimitInterceptor interceptor = new RateLimitInterceptor(1, 100); + IngestContext first = new IngestContext("trace-1"); + IngestContext second = new IngestContext("trace-2"); + + boolean firstAllowed = interceptor.before(frame("unknown", "13079970001"), first); + boolean secondAllowed = interceptor.before(frame("unknown", "13079970001"), second); + + assertThat(firstAllowed).isTrue(); + assertThat(secondAllowed).isFalse(); + assertThat(second.aborted()).isTrue(); + assertThat(second.abortReason()).isEqualTo("rate-limited:phone:13079970001"); + } + + private static RawFrame frame(String vin, String phone) { + return new RawFrame( + ProtocolId.JT808, + 0x0200, + 0, + new Object(), + new byte[]{0x01}, + Map.of("vin", vin, "phone", phone), + Instant.parse("2026-06-29T00:00:00Z")); + } +} diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java index 5c1ba757..afaad22c 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java @@ -25,16 +25,45 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor { private final EventFileStore store; private final TelemetryEnvelopeRecordMapper mapper; private final TdengineHistoryWriter tdengineWriter; + private final boolean telemetryFieldsEnabled; public EventHistoryEnvelopeIngestor(EventFileStore store, TelemetryEnvelopeRecordMapper mapper) { this(store, mapper, null); } + public EventHistoryEnvelopeIngestor(TelemetryEnvelopeRecordMapper mapper, + TdengineHistoryWriter tdengineWriter) { + this(mapper, tdengineWriter, true); + } + + public EventHistoryEnvelopeIngestor(TelemetryEnvelopeRecordMapper mapper, + TdengineHistoryWriter tdengineWriter, + boolean telemetryFieldsEnabled) { + this(null, mapper, tdengineWriter, false, telemetryFieldsEnabled); + } + public EventHistoryEnvelopeIngestor(EventFileStore store, TelemetryEnvelopeRecordMapper mapper, TdengineHistoryWriter tdengineWriter) { + this(store, mapper, tdengineWriter, true, true); + } + + public EventHistoryEnvelopeIngestor(EventFileStore store, + TelemetryEnvelopeRecordMapper mapper, + TdengineHistoryWriter tdengineWriter, + boolean telemetryFieldsEnabled) { + this(store, mapper, tdengineWriter, true, telemetryFieldsEnabled); + } + + private EventHistoryEnvelopeIngestor(EventFileStore store, + TelemetryEnvelopeRecordMapper mapper, + TdengineHistoryWriter tdengineWriter, + boolean requireStore, + boolean telemetryFieldsEnabled) { if (store == null) { - throw new IllegalArgumentException("store must not be null"); + if (requireStore) { + throw new IllegalArgumentException("store must not be null"); + } } if (mapper == null) { throw new IllegalArgumentException("mapper must not be null"); @@ -42,12 +71,15 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor { this.store = store; this.mapper = mapper; this.tdengineWriter = tdengineWriter; + this.telemetryFieldsEnabled = telemetryFieldsEnabled; } public void ingest(byte[] kafkaValue) throws IOException { VehicleEnvelope envelope = parse(kafkaValue); - EventFileRecord record = mapper.toRecord(envelope); - store.append(record); + if (store != null) { + EventFileRecord record = mapper.toRecord(envelope); + store.append(record); + } writeTdengineFacts(List.of(envelope)); } @@ -67,7 +99,7 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor { VehicleEnvelope envelope = null; try { envelope = parse(kafkaValue); - EventFileRecord record = mapper.toRecord(envelope); + EventFileRecord record = store == null ? null : mapper.toRecord(envelope); results.add(null); valid.add(new BatchEntry(results.size() - 1, envelope, record)); } catch (IllegalArgumentException ex) { @@ -80,12 +112,14 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor { return List.copyOf(results); } try { - store.appendAll(valid.stream().map(BatchEntry::record).toList()); + if (store != null) { + store.appendAll(valid.stream().map(BatchEntry::record).toList()); + } writeTdengineFacts(valid.stream().map(BatchEntry::envelope).toList()); for (BatchEntry entry : valid) { results.set(entry.index(), EnvelopeIngestResult.stored( - entry.record().eventId(), - entry.record().vin())); + entry.eventId(), + entry.vin())); } } catch (IOException ex) { for (BatchEntry entry : valid) { @@ -99,6 +133,13 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor { } private record BatchEntry(int index, VehicleEnvelope envelope, EventFileRecord record) { + private String eventId() { + return record == null ? envelope.getEventId() : record.eventId(); + } + + private String vin() { + return record == null ? envelope.getVin() : record.vin(); + } } private static VehicleEnvelope parse(byte[] kafkaValue) { @@ -128,11 +169,13 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor { if (!locations.isEmpty()) { tdengineWriter.appendLocations(locations); } - var telemetryFields = envelopes.stream() - .flatMap(envelope -> TdengineEnvelopeRows.telemetryFields(envelope).stream()) - .toList(); - if (!telemetryFields.isEmpty()) { - tdengineWriter.appendTelemetryFields(telemetryFields); + if (telemetryFieldsEnabled) { + var telemetryFields = envelopes.stream() + .flatMap(envelope -> TdengineEnvelopeRows.telemetryFields(envelope).stream()) + .toList(); + if (!telemetryFields.isEmpty()) { + tdengineWriter.appendTelemetryFields(telemetryFields); + } } } } 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 f91e7f99..13acf47b 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 @@ -186,6 +186,42 @@ class EventHistoryEnvelopeIngestorTest { assertThat(tdengineWriter.telemetryFields.getFirst().valueDouble()).isEqualTo(42.5); } + @Test + void tryIngestCanWriteTdengineFactsWithoutEventFileStore() { + CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter(); + EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor( + new TelemetryEnvelopeRecordMapper(), tdengineWriter); + + EnvelopeIngestResult result = ingestor.tryIngest( + jt808LocationEnvelope("jt808-location-1", "frame-jt808-1", "013800000001").toByteArray()); + + assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED); + assertThat(tdengineWriter.rawFrames).extracting(TdengineRawFrameRow::frameId) + .containsExactly("frame-jt808-1"); + assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId) + .containsExactly("jt808-location-1"); + assertThat(tdengineWriter.telemetryFields) + .extracting(TdengineTelemetryFieldRow::fieldKey) + .containsExactly("location.speedKmh"); + } + + @Test + void tryIngestCanSkipTelemetryFieldFactsForHighThroughputRuntime() { + CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter(); + EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor( + new TelemetryEnvelopeRecordMapper(), tdengineWriter, false); + + EnvelopeIngestResult result = ingestor.tryIngest( + jt808LocationEnvelope("jt808-location-1", "frame-jt808-1", "013800000001").toByteArray()); + + assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED); + assertThat(tdengineWriter.rawFrames).extracting(TdengineRawFrameRow::frameId) + .containsExactly("frame-jt808-1"); + assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId) + .containsExactly("jt808-location-1"); + assertThat(tdengineWriter.telemetryFields).isEmpty(); + } + @Test void tryIngestAllBatchesFileStoreAndTdengineWrites() throws Exception { CapturingStore store = new CapturingStore(); diff --git a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactory.java b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactory.java index e0123cda..7d8a82cd 100644 --- a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactory.java +++ b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactory.java @@ -44,11 +44,15 @@ public final class KafkaEnvelopeConsumerFactory { if (topics.isEmpty()) { continue; } - KafkaEnvelopeConsumerWorker worker = new KafkaEnvelopeConsumerWorker( - consumerFactory.apply(consumerProperties(props, binding, processorBeanName)), - topicProcessors(topics, processor)); - worker.subscribe(topics); - workers.add(worker); + int concurrency = Math.max(1, props.getConsumer().getConcurrency()); + for (int workerIndex = 0; workerIndex < concurrency; workerIndex++) { + KafkaEnvelopeConsumerWorker worker = new KafkaEnvelopeConsumerWorker( + consumerFactory.apply(consumerProperties(props, binding, processorBeanName, + workerIndex, concurrency)), + topicProcessors(topics, processor)); + worker.subscribe(topics); + workers.add(worker); + } } return workers; } @@ -104,14 +108,17 @@ public final class KafkaEnvelopeConsumerFactory { private Properties consumerProperties(SinkMqProperties props, SinkMqProperties.Binding binding, - String processorBeanName) { + String processorBeanName, + int workerIndex, + int concurrency) { SinkMqProperties.Consumer consumer = props.getConsumer(); Properties p = new Properties(); p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers()); p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); p.put(ConsumerConfig.GROUP_ID_CONFIG, groupId(binding, processorBeanName)); - p.put(ConsumerConfig.CLIENT_ID_CONFIG, consumer.getClientIdPrefix() + "-" + processorBeanName); + p.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId(consumer.getClientIdPrefix(), processorBeanName, + workerIndex, concurrency)); // 手动提交 offset：只有本批次至少有一条记录被处理器接收后才 commit， // 避免轮询到空批次或未绑定 topic 时推进消费位点。 p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); @@ -121,6 +128,11 @@ public final class KafkaEnvelopeConsumerFactory { return p; } + private String clientId(String prefix, String processorBeanName, int workerIndex, int concurrency) { + String base = prefix + "-" + processorBeanName; + return concurrency <= 1 ? base : base + "-" + workerIndex; + } + private String groupId(SinkMqProperties.Binding binding, String processorBeanName) { if (binding.getGroupId() != null && !binding.getGroupId().isBlank()) { return binding.getGroupId(); 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 bf590dde..5026004f 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 @@ -63,7 +63,7 @@ public final class KafkaEventSink implements EventSink, AutoCloseable { } String topic = breaker.tryAcquirePermission() ? router.route(event) : dlqTopic; - ProducerRecord record = new ProducerRecord<>(topic, event.vin(), payload); + ProducerRecord record = new ProducerRecord<>(topic, partitionKey(event), payload); record.headers().add("event-id", event.eventId().getBytes()); record.headers().add("trace-id", event.traceId() == null ? new byte[0] : event.traceId().getBytes()); record.headers().add("source", event.source().name().getBytes()); @@ -80,6 +80,36 @@ public final class KafkaEventSink implements EventSink, AutoCloseable { return cf; } + private static String partitionKey(VehicleEvent event) { + String vin = event.vin(); + if (isKnown(vin)) { + return vin; + } + String vehicleKey = metadata(event, "vehicleKey"); + if (!isKnown(vehicleKey)) { + vehicleKey = metadata(event, "vehicle_key"); + } + if (isKnown(vehicleKey)) { + return "vehicleKey:" + vehicleKey; + } + String phone = metadata(event, "phone"); + if (isKnown(phone)) { + return "phone:" + phone; + } + return vin == null || vin.isBlank() ? "unknown" : vin; + } + + private static String metadata(VehicleEvent event, String key) { + if (event.metadata() == null) { + return ""; + } + return event.metadata().getOrDefault(key, "").trim(); + } + + private static boolean isKnown(String value) { + return value != null && !value.isBlank() && !"unknown".equalsIgnoreCase(value.trim()); + } + @Override public void close() { producer.flush(); 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 92aa9644..b7b16299 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 @@ -94,6 +94,7 @@ public class SinkMqProperties { private String autoOffsetReset = "earliest"; private int maxPollRecords = 500; private int maxPollIntervalMillis = 1800000; + private int concurrency = 1; /** * Processor bean name -> Kafka binding。 * @@ -118,6 +119,8 @@ public class SinkMqProperties { public void setMaxPollRecords(int maxPollRecords) { this.maxPollRecords = maxPollRecords; } public int getMaxPollIntervalMillis() { return maxPollIntervalMillis; } public void setMaxPollIntervalMillis(int maxPollIntervalMillis) { this.maxPollIntervalMillis = maxPollIntervalMillis; } + public int getConcurrency() { return concurrency; } + public void setConcurrency(int concurrency) { this.concurrency = concurrency; } public Map getBindings() { return bindings; } public void setBindings(Map bindings) { this.bindings = bindings; } } diff --git a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactoryTest.java b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactoryTest.java index c1117ab2..7a4a286f 100644 --- a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactoryTest.java +++ b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactoryTest.java @@ -44,6 +44,32 @@ class KafkaEnvelopeConsumerFactoryTest { }); } + @Test + void createsParallelConsumersForEachBindingWhenConsumerConcurrencyIsConfigured() { + List created = new ArrayList<>(); + KafkaEnvelopeConsumerFactory factory = new KafkaEnvelopeConsumerFactory(props -> { + created.add(props); + return new MockConsumer<>(OffsetResetStrategy.EARLIEST); + }); + SinkMqProperties props = new SinkMqProperties(); + props.getConsumer().setConcurrency(3); + EnvelopeConsumerProcessor stateProcessor = processor(); + + List workers = factory.createWorkers(Map.of( + "vehicleStateEnvelopeConsumerProcessor", stateProcessor), props); + + assertThat(workers).hasSize(3); + assertThat(created) + .extracting(p -> p.getProperty(ConsumerConfig.GROUP_ID_CONFIG)) + .containsExactly("vehicle-state", "vehicle-state", "vehicle-state"); + assertThat(created) + .extracting(p -> p.getProperty(ConsumerConfig.CLIENT_ID_CONFIG)) + .containsExactly( + "lingniu-envelope-consumer-vehicleStateEnvelopeConsumerProcessor-0", + "lingniu-envelope-consumer-vehicleStateEnvelopeConsumerProcessor-1", + "lingniu-envelope-consumer-vehicleStateEnvelopeConsumerProcessor-2"); + } + private EnvelopeConsumerProcessor processor() { return new EnvelopeConsumerProcessor( "service", 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 index c85154a2..4fbdd843 100644 --- 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 @@ -1,14 +1,23 @@ package com.lingniu.ingest.sink.mq; import com.lingniu.ingest.api.ProtocolId; +import com.lingniu.ingest.api.event.LocationPayload; import com.lingniu.ingest.api.event.VehicleEvent; import io.github.resilience4j.circuitbreaker.CircuitBreaker; +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import java.time.Instant; import java.util.Map; +import java.util.concurrent.CompletableFuture; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; class KafkaEventSinkTest { @@ -24,6 +33,28 @@ class KafkaEventSinkTest { assertThat(sink.accepts(rawArchive())).isTrue(); } + @Test + void usesPhoneAsKafkaKeyWhenVinIsUnknown() { + @SuppressWarnings("unchecked") + KafkaProducer producer = mock(KafkaProducer.class); + @SuppressWarnings("unchecked") + ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); + when(producer.send(sent.capture(), any())).thenAnswer(invocation -> { + invocation.getArgument(1).onCompletion(null, null); + return CompletableFuture.completedFuture(null); + }); + KafkaEventSink sink = new KafkaEventSink( + producer, + new EnvelopeMapper("node-1"), + new TopicRouter(new SinkMqProperties.Topics()), + "vehicle.dlq.gb32960.v1", + CircuitBreaker.ofDefaults("kafka-test")); + + sink.publish(jt808Location()).join(); + + assertThat(sent.getValue().key()).isEqualTo("phone:13079979000"); + } + private static VehicleEvent.RawArchive rawArchive() { return new VehicleEvent.RawArchive( "raw-1", @@ -37,4 +68,16 @@ class KafkaEventSinkTest { 0, new byte[] {0x23, 0x23}); } + + private static VehicleEvent.Location jt808Location() { + return new VehicleEvent.Location( + "loc-1", + "unknown", + ProtocolId.JT808, + Instant.parse("2026-06-29T11:22:00Z"), + Instant.parse("2026-06-29T11:22:01Z"), + "trace-loc-1", + Map.of("phone", "13079979000"), + new LocationPayload(121.1, 31.2, 8.0, 40.0, 90.0, 0L, 1L, null)); + } } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java index 008926af..a9166808 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java @@ -3,7 +3,6 @@ package com.lingniu.ingest.tdenginehistory; import javax.sql.DataSource; import java.io.IOException; import java.sql.Connection; -import java.sql.PreparedStatement; import java.sql.SQLException; import java.sql.Statement; import java.sql.Timestamp; @@ -13,21 +12,19 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { - private static final Logger log = LoggerFactory.getLogger(TdengineJdbcHistoryWriter.class); private static final int MAX_LITERAL_ROWS_PER_STATEMENT = 200; private final DataSource dataSource; private final TdengineHistorySchema schema; private final TdengineHistoryStatements statements; private final Object schemaInitializationMonitor = new Object(); + private final Map lastTimestampByInsertSql = new ConcurrentHashMap<>(); private volatile boolean schemaInitialized; - private volatile boolean literalFallbackLogged; public TdengineJdbcHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) { if (dataSource == null) { @@ -95,87 +92,42 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { } } Set createdChildTables = new LinkedHashSet<>(); - Map preparedStatements = new LinkedHashMap<>(); Map literalBatches = new LinkedHashMap<>(); - Map literalOnlyBatches = new LinkedHashMap<>(); - try { - for (T row : rows) { - if (row == null) { - continue; - } - TdengineBatchStatement batch = mapper.apply(row); - if (createdChildTables.add(batch.createChildTableSql())) { - try (Statement statement = connection.createStatement()) { - statement.execute(batch.createChildTableSql()); - } - } - if (hasEmptyString(batch.values())) { - literalOnlyBatches.computeIfAbsent(batch.insertSql(), LiteralBatch::new).add(batch.values()); - continue; - } - PreparedStatement prepared = preparedStatements.computeIfAbsent(batch.insertSql(), sql -> { - try { - return connection.prepareStatement(sql); - } catch (SQLException e) { - throw new JdbcRuntimeException(e); - } - }); - bind(prepared, batch.values()); - prepared.addBatch(); - literalBatches.computeIfAbsent(batch.insertSql(), LiteralBatch::new).add(batch.values()); + for (T row : rows) { + if (row == null) { + continue; } - for (Map.Entry entry : preparedStatements.entrySet()) { - try { - entry.getValue().executeBatch(); - } catch (SQLException e) { - executeLiteralFallback(connection, literalBatches.get(entry.getKey()), e); + TdengineBatchStatement batch = mapper.apply(row); + if (createdChildTables.add(batch.createChildTableSql())) { + try (Statement statement = connection.createStatement()) { + statement.execute(batch.createChildTableSql()); } } - for (LiteralBatch batch : literalOnlyBatches.values()) { - executeLiteralBatch(connection, batch); - } - } finally { - for (PreparedStatement prepared : preparedStatements.values()) { - prepared.close(); - } + literalBatches.computeIfAbsent(batch.insertSql(), LiteralBatch::new) + .add(uniqueTimestampValues(batch.insertSql(), batch.values())); + } + for (LiteralBatch batch : literalBatches.values()) { + executeLiteralBatch(connection, batch); } }); } - private static void bind(PreparedStatement prepared, List values) throws SQLException { - for (int i = 0; i < values.size(); i++) { - Object value = values.get(i); - int parameterIndex = i + 1; - if (value instanceof Instant instant) { - prepared.setTimestamp(parameterIndex, Timestamp.from(instant)); - } else { - prepared.setObject(parameterIndex, value); + private List uniqueTimestampValues(String insertSql, List values) { + if (values == null || values.isEmpty() || !(values.getFirst() instanceof Instant candidate)) { + return values; + } + Instant uniqueTimestamp = lastTimestampByInsertSql.compute(insertSql, (key, lastTimestamp) -> { + if (lastTimestamp == null || candidate.isAfter(lastTimestamp)) { + return candidate; } + return lastTimestamp.plusMillis(1); + }); + if (candidate.equals(uniqueTimestamp)) { + return values; } - } - - private static boolean hasEmptyString(List values) { - for (Object value : values) { - if (value instanceof String stringValue && stringValue.isEmpty()) { - return true; - } - } - return false; - } - - private void executeLiteralFallback(Connection connection, - LiteralBatch batch, - SQLException preparedFailure) throws SQLException { - if (batch == null || batch.rows().isEmpty()) { - throw preparedFailure; - } - if (!literalFallbackLogged) { - literalFallbackLogged = true; - log.warn("TDengine prepared batch failed; falling back to literal inserts: {}", - preparedFailure.getMessage()); - log.debug("TDengine prepared batch failure stacktrace", preparedFailure); - } - executeLiteralBatch(connection, batch); + List copy = new java.util.ArrayList<>(values); + copy.set(0, uniqueTimestamp); + return copy; } private static void executeLiteralBatch(Connection connection, LiteralBatch batch) throws SQLException { diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java index f9e1823b..10f18269 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java @@ -46,13 +46,13 @@ class TdengineJdbcHistoryWriterTest { assertThat(jdbc.executedSql) .filteredOn(sql -> sql.contains("USING raw_frames TAGS")) .hasSize(1); - assertThat(jdbc.preparedBatches).hasSize(1); - RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next(); - assertThat(batch.sql()).contains("INSERT INTO raw_jt808_"); - assertThat(batch.rows()).hasSize(2); - assertThat(batch.rows().getFirst().get(1)).isEqualTo(Timestamp.from(first.ts())); - assertThat(batch.rows().getFirst().get(2)).isEqualTo("frame-1"); - assertThat(batch.rows().get(1).get(2)).isEqualTo("frame-2"); + assertThat(jdbc.preparedBatches).isEmpty(); + assertThat(jdbc.executedSql) + .anyMatch(sql -> sql.startsWith("INSERT INTO raw_jt808_") + && sql.contains(Long.toString(first.ts().toEpochMilli())) + && sql.contains("frame-1") + && sql.contains("frame-2") + && sql.contains(") (")); assertThat(jdbc.commits).isEqualTo(1); } @@ -85,9 +85,12 @@ class TdengineJdbcHistoryWriterTest { assertThat(jdbc.executedSql).startsWith(schema.bootstrapSql().toArray(String[]::new)); assertThat(jdbc.executedSql) .anyMatch(sql -> sql.contains("USING vehicle_locations TAGS")); - RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next(); - assertThat(batch.sql()).contains("INSERT INTO loc_gb32960_"); - assertThat(batch.rows().getFirst().get(12)).isNull(); + assertThat(jdbc.preparedBatches).isEmpty(); + assertThat(jdbc.executedSql) + .anyMatch(sql -> sql.startsWith("INSERT INTO loc_gb32960_") + && sql.contains("fact-1") + && sql.contains("NULL") + && sql.contains("archive://gb32960/frame-1.bin")); } @Test @@ -103,12 +106,13 @@ class TdengineJdbcHistoryWriterTest { assertThat(jdbc.executedSql) .filteredOn(sql -> sql.contains("USING telemetry_fields TAGS")) .hasSize(1); - RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next(); - assertThat(batch.sql()).contains("INSERT INTO tf_gb32960_"); - assertThat(batch.rows()).hasSize(2); - assertThat(batch.rows().getFirst().get(2)).isEqualTo("evt-1#0"); - assertThat(batch.rows().getFirst().get(7)).isEqualTo(123.4); - assertThat(batch.rows().get(1).get(6)).isEqualTo("124.5"); + assertThat(jdbc.preparedBatches).isEmpty(); + assertThat(jdbc.executedSql) + .anyMatch(sql -> sql.startsWith("INSERT INTO tf_gb32960_") + && sql.contains("evt-1#0") + && sql.contains("evt-2#0") + && sql.contains("123.4") + && sql.contains("'124.5'")); } @Test @@ -126,14 +130,14 @@ class TdengineJdbcHistoryWriterTest { } @Test - void fallsBackToLiteralInsertWhenPreparedBatchFails() throws Exception { + void writesLiteralInsertWithoutPreparedBatch() throws Exception { RecordingJdbc jdbc = new RecordingJdbc(); jdbc.failPreparedBatches = true; TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema); writer.appendRawFrames(List.of(rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z"), "none"))); - assertThat(jdbc.preparedBatches).hasSize(1); + assertThat(jdbc.preparedBatches).isEmpty(); assertThat(jdbc.executedSql) .anyMatch(sql -> sql.startsWith("INSERT INTO raw_jt808_") && sql.contains("parse_error") @@ -175,6 +179,25 @@ class TdengineJdbcHistoryWriterTest { .contains(") ("); } + @Test + void bumpsDuplicateTimestampsForSameChildTable() throws Exception { + RecordingJdbc jdbc = new RecordingJdbc(); + TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema); + Instant ts = Instant.parse("2026-06-29T05:00:01Z"); + + writer.appendRawFrames(List.of( + rawFrame("frame-1", ts), + rawFrame("frame-2", ts))); + + String insert = jdbc.executedSql.stream() + .filter(sql -> sql.startsWith("INSERT INTO raw_jt808_")) + .findFirst() + .orElseThrow(); + assertThat(insert) + .contains("(" + ts.toEpochMilli() + ", 'frame-1'") + .contains("(" + (ts.toEpochMilli() + 1) + ", 'frame-2'"); + } + private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) { return rawFrame(frameId, ts, ""); } diff --git a/tools/jt808_e2e_smoke.py b/tools/jt808_e2e_smoke.py index 0ca2990b..c7e7116d 100755 --- a/tools/jt808_e2e_smoke.py +++ b/tools/jt808_e2e_smoke.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import concurrent.futures import datetime as dt import json import os @@ -12,6 +13,7 @@ import socket import time import urllib.parse import urllib.request +from dataclasses import dataclass from typing import Any try: @@ -220,6 +222,83 @@ def send_phone_sequence( return ack, sent +@dataclass(frozen=True) +class SendWorkloadResult: + sent_registers: int + sent_locations: int + elapsed_seconds: float + + +def send_one_phone(args: argparse.Namespace, + phone: str, + phone_index: int, + event_time: dt.datetime) -> tuple[bytes, int]: + if args.connection_mode == "session": + return send_phone_sequence( + args.tcp_host, + args.tcp_port, + phone, + phone_index, + args.frames, + event_time, + args.tcp_timeout, + args.rate, + args.linger_ms, + args.post_register_ms, + ) + + ack = send_register(args.tcp_host, args.tcp_port, phone, 100 + phone_index, args.tcp_timeout) + if args.post_register_ms > 0: + time.sleep(args.post_register_ms / 1000) + sent = 0 + for frame_index in range(args.frames): + serial = 1000 + phone_index * args.frames + frame_index + send_location( + args.tcp_host, + args.tcp_port, + phone, + serial, + event_time + dt.timedelta(seconds=frame_index), + args.tcp_timeout, + args.linger_ms, + ) + sent += 1 + if args.rate > 0: + time.sleep(1 / args.rate) + return ack, sent + + +def send_phone_workload(args: argparse.Namespace, + phones: list[str], + event_time: dt.datetime) -> SendWorkloadResult: + sent_locations = 0 + start = time.monotonic() + workers = max(1, int(getattr(args, "workers", 1))) + + def run_one(index_and_phone: tuple[int, str]) -> tuple[bytes, int]: + phone_index, phone = index_and_phone + return send_one_phone(args, phone, phone_index, event_time) + + indexed_phones = list(enumerate(phones)) + if workers == 1 or len(indexed_phones) <= 1: + results = [run_one(item) for item in indexed_phones] + else: + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: + results = list(executor.map(run_one, indexed_phones)) + + for phone, (ack, sent) in zip(phones, results): + if args.require_register_ack and not ack: + raise RuntimeError(f"register ack timeout for {phone}") + sent_locations += sent + + elapsed = max(time.monotonic() - start, 0.001) + return SendWorkloadResult( + sent_registers=len(phones), + sent_locations=sent_locations, + elapsed_seconds=elapsed, + ) + + def query_json(base_url: str, path: str, params: dict[str, Any], timeout: float) -> dict[str, Any]: url = base_url.rstrip("/") + path + "?" + urllib.parse.urlencode(params) request = urllib.request.Request(url, headers={"Accept": "application/json"}) @@ -286,7 +365,7 @@ def verify_tdengine_raw_frames( request_timeout: float, ) -> int: if not sqls: - raise RuntimeError("no raw uri available for tdengine raw_frames verification") + return 0 verified = 0 for sql in sqls: count = wait_for_tdengine_raw_frames( @@ -366,49 +445,7 @@ def run(args: argparse.Namespace) -> dict[str, Any]: date_from = (event_time - dt.timedelta(minutes=5)).isoformat() date_to = (event_time + dt.timedelta(minutes=args.frames + 5)).isoformat() phones = [generated_phone(args.start_phone, index) for index in range(args.phones)] - sent_locations = 0 - sent_registers = 0 - start = time.monotonic() - - for phone_index, phone in enumerate(phones): - if args.connection_mode == "session": - ack, sent = send_phone_sequence( - args.tcp_host, - args.tcp_port, - phone, - phone_index, - args.frames, - event_time, - args.tcp_timeout, - args.rate, - args.linger_ms, - args.post_register_ms, - ) - else: - ack = send_register(args.tcp_host, args.tcp_port, phone, 100 + phone_index, args.tcp_timeout) - if args.post_register_ms > 0: - time.sleep(args.post_register_ms / 1000) - sent = 0 - for frame_index in range(args.frames): - serial = 1000 + phone_index * args.frames + frame_index - send_location( - args.tcp_host, - args.tcp_port, - phone, - serial, - event_time + dt.timedelta(seconds=frame_index), - args.tcp_timeout, - args.linger_ms, - ) - sent += 1 - if args.rate > 0: - time.sleep(1 / args.rate) - sent_registers += 1 - if args.require_register_ack and not ack: - raise RuntimeError(f"register ack timeout for {phone}") - sent_locations += sent - - elapsed = max(time.monotonic() - start, 0.001) + workload = send_phone_workload(args, phones, event_time) first_phone = phones[0] page = wait_for_locations( args.history_base_url, @@ -453,11 +490,12 @@ def run(args: argparse.Namespace) -> dict[str, Any]: "tcpHost": args.tcp_host, "tcpPort": args.tcp_port, "connectionMode": args.connection_mode, + "workers": max(1, int(args.workers)), "phones": phones, - "sentRegisters": sent_registers, - "sentLocations": sent_locations, - "sendElapsedSeconds": round(elapsed, 3), - "sendLocationsPerSecond": round(sent_locations / elapsed, 2), + "sentRegisters": workload.sent_registers, + "sentLocations": workload.sent_locations, + "sendElapsedSeconds": round(workload.elapsed_seconds, 3), + "sendLocationsPerSecond": round(workload.sent_locations / workload.elapsed_seconds, 2), "historyRowsForFirstPhone": len(items), "tdengineRawFrames": tdengine_raw_frames, "archiveChecked": archive_checked, @@ -475,6 +513,7 @@ def parser() -> argparse.ArgumentParser: p.add_argument("--connection-mode", choices=["per-frame", "session"], default="per-frame") p.add_argument("--start-phone", default="13079962000") p.add_argument("--phones", type=int, default=1) + p.add_argument("--workers", type=int, default=1, help="parallel phone sessions; 1 keeps legacy serial behavior") p.add_argument("--frames", type=int, default=3, help="location frames per phone") p.add_argument("--rate", type=float, default=0, help="location send rate per second; 0 means unlimited") p.add_argument("--linger-ms", type=int, default=2500, help="time to keep each terminal connection open after writes") diff --git a/tools/test_jt808_e2e_smoke.py b/tools/test_jt808_e2e_smoke.py index c01deaca..9ed59263 100644 --- a/tools/test_jt808_e2e_smoke.py +++ b/tools/test_jt808_e2e_smoke.py @@ -1,6 +1,7 @@ import pathlib import types import unittest +from unittest import mock from tools import jt808_e2e_smoke as smoke @@ -70,11 +71,64 @@ class Jt808E2ESmokeTest(unittest.TestCase): self.assertTrue(sqls[0].endswith("frame-1.bin'")) self.assertTrue(sqls[1].endswith("frame-2.bin'")) + def test_verify_tdengine_raw_frames_allows_probe_without_visible_raw_uri(self): + verified = smoke.verify_tdengine_raw_frames( + "http://localhost:6041/rest/sql/vehicle_ts", + "root", + "taosdata", + [], + 0, + 1, + ) + + self.assertEqual(verified, 0) + def test_expected_history_count_waits_for_pagination_sample(self): args = types.SimpleNamespace(frames=3, expect_history_count=1, verify_pagination=True) self.assertEqual(smoke.expected_history_count(args), 2) + def test_send_phone_workload_can_run_session_phones_with_workers(self): + event_time = smoke.parse_event_time("2024-01-02T03:04:05") + args = types.SimpleNamespace( + connection_mode="session", + tcp_host="127.0.0.1", + tcp_port=808, + frames=3, + tcp_timeout=1, + rate=0, + linger_ms=0, + post_register_ms=0, + require_register_ack=True, + workers=2, + ) + calls = [] + + def fake_send_phone_sequence(host, port, phone, phone_index, frames, sent_event_time, + timeout, rate, linger_ms, post_register_ms): + calls.append((host, port, phone, phone_index, frames, sent_event_time, + timeout, rate, linger_ms, post_register_ms)) + return b"ack", frames + + with mock.patch.object(smoke, "send_phone_sequence", side_effect=fake_send_phone_sequence): + result = smoke.send_phone_workload( + args, + ["13079962000", "13079962001", "13079962002"], + event_time, + ) + + self.assertEqual(result.sent_registers, 3) + self.assertEqual(result.sent_locations, 9) + self.assertGreater(result.elapsed_seconds, 0) + self.assertEqual( + sorted((call[2], call[3], call[4]) for call in calls), + [ + ("13079962000", 0, 3), + ("13079962001", 1, 3), + ("13079962002", 2, 3), + ], + ) + if __name__ == "__main__": unittest.main()