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

fix: harden jt808 tdengine ingestion
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 19:53:08 +08:00
parent d1748fcc2f
commit 12de83e37f
17 changed files with 588 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactory.java
View File

@@ -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();

32
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java
View File

@@ -63,7 +63,7 @@ public final class KafkaEventSink implements EventSink, AutoCloseable {
}
String topic = breaker.tryAcquirePermission() ? router.route(event) : dlqTopic;
ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, event.vin(), payload);
ProducerRecord<String, byte[]> 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();

3
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java
View File

@@ -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<String, Binding> getBindings() { return bindings; }
public void setBindings(Map<String, Binding> bindings) { this.bindings = bindings; }
}

26
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerFactoryTest.java
View File

@@ -44,6 +44,32 @@ class KafkaEnvelopeConsumerFactoryTest {
});
}
@Test
void createsParallelConsumersForEachBindingWhenConsumerConcurrencyIsConfigured() {
List<Properties> 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<KafkaEnvelopeConsumerWorker> 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",

43
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java
View File

@@ -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<String, byte[]> producer = mock(KafkaProducer.class);
@SuppressWarnings("unchecked")
ArgumentCaptor<ProducerRecord<String, byte[]>> sent = ArgumentCaptor.forClass(ProducerRecord.class);
when(producer.send(sent.capture(), any())).thenAnswer(invocation -> {
invocation.<Callback>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));
}
}

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

@@ -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<String, Instant> 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<String> createdChildTables = new LinkedHashSet<>();
Map<String, PreparedStatement> preparedStatements = new LinkedHashMap<>();
Map<String, LiteralBatch> literalBatches = new LinkedHashMap<>();
Map<String, LiteralBatch> 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<String, PreparedStatement> 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<Object> 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<Object> uniqueTimestampValues(String insertSql, List<Object> 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<Object> 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<Object> copy = new java.util.ArrayList<>(values);
copy.set(0, uniqueTimestamp);
return copy;
}
private static void executeLiteralBatch(Connection connection, LiteralBatch batch) throws SQLException {

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

@@ -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, "");
}