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

fix: keep jt808 raw history consumer current

Browse Source
This commit is contained in:
lingniu
2026-06-29 15:56:41 +08:00
parent 57ef82f563
commit 98b37e71e9
13 changed files with 274 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -117,6 +117,7 @@ public final class KafkaEnvelopeConsumerFactory {
p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumer.getAutoOffsetReset());
p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, consumer.getMaxPollRecords());
p.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, consumer.getMaxPollIntervalMillis());
return p;
}

17
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerWorker.java
View File

@@ -7,7 +7,10 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
public final class KafkaEnvelopeConsumerWorker implements AutoCloseable {
@@ -33,22 +36,26 @@ public final class KafkaEnvelopeConsumerWorker implements AutoCloseable {
public int pollOnce(Duration timeout) {
ConsumerRecords<String, byte[]> records = consumer.poll(timeout == null ? Duration.ZERO : timeout);
int processed = 0;
Map<EnvelopeConsumerProcessor, List<EnvelopeConsumerRecord>> batches = new LinkedHashMap<>();
for (ConsumerRecord<String, byte[]> record : records) {
EnvelopeConsumerProcessor processor = processorsByTopic.get(record.topic());
if (processor == null) {
// worker 可能订阅多个 topic；没有显式绑定处理器的 topic 不参与提交语义。
continue;
}
// EnvelopeConsumerProcessor 内部会把解析或业务错误转成 DLQ 记录，
// 这里保持 Kafka worker 的职责单一：轮询、分发、成功后提交 offset。
processor.process(new EnvelopeConsumerRecord(
batches.computeIfAbsent(processor, ignored -> new ArrayList<>()).add(new EnvelopeConsumerRecord(
record.topic(),
record.partition(),
record.offset(),
record.key(),
record.value()));
processed++;
}
int processed = 0;
for (Map.Entry<EnvelopeConsumerProcessor, List<EnvelopeConsumerRecord>> entry : batches.entrySet()) {
// EnvelopeConsumerProcessor 内部会把解析或业务错误转成 DLQ 记录，
// 这里保持 Kafka worker 的职责单一：轮询、批量分发、成功后提交 offset。
entry.getKey().processAll(entry.getValue());
processed += entry.getValue().size();
}
if (processed > 0) {
// commitSync 放在批次末尾，保证同一个 poll 批次内的消息按 Kafka offset 一起确认。

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

@@ -93,6 +93,7 @@ public class SinkMqProperties {
private int loopBackoffMillis = 1000;
private String autoOffsetReset = "earliest";
private int maxPollRecords = 500;
private int maxPollIntervalMillis = 1800000;
/**
* Processor bean name -> Kafka binding。
*
@@ -115,6 +116,8 @@ public class SinkMqProperties {
public void setAutoOffsetReset(String autoOffsetReset) { this.autoOffsetReset = autoOffsetReset; }
public int getMaxPollRecords() { return maxPollRecords; }
public void setMaxPollRecords(int maxPollRecords) { this.maxPollRecords = maxPollRecords; }
public int getMaxPollIntervalMillis() { return maxPollIntervalMillis; }
public void setMaxPollIntervalMillis(int maxPollIntervalMillis) { this.maxPollIntervalMillis = maxPollIntervalMillis; }
public Map<String, Binding> getBindings() { return bindings; }
public void setBindings(Map<String, Binding> bindings) { this.bindings = bindings; }
}

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

@@ -40,6 +40,7 @@ class KafkaEnvelopeConsumerFactoryTest {
.allSatisfy(p -> {
assertThat(p.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)).isEqualTo("kafka-1:9092");
assertThat(p.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isEqualTo(false);
assertThat(p.get(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)).isEqualTo(1800000);
});
}

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

@@ -20,6 +20,7 @@ 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;
@@ -179,26 +180,39 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter {
private static void executeLiteralBatch(Connection connection, LiteralBatch batch) throws SQLException {
try (Statement statement = connection.createStatement()) {
for (List<Object> row : batch.rows()) {
statement.execute(toLiteralInsertSql(batch.insertSql(), row));
List<List<Object>> rows = batch.rows();
for (int start = 0; start < rows.size(); start += MAX_LITERAL_ROWS_PER_STATEMENT) {
int end = Math.min(start + MAX_LITERAL_ROWS_PER_STATEMENT, rows.size());
statement.execute(toLiteralInsertSql(batch.insertSql(), rows.subList(start, end)));
}
}
}
private static String toLiteralInsertSql(String insertSql, List<Object> values) {
private static String toLiteralInsertSql(String insertSql, List<List<Object>> rows) {
int valuesIndex = insertSql.lastIndexOf("VALUES");
if (valuesIndex < 0) {
throw new IllegalArgumentException("insert SQL must contain VALUES: " + insertSql);
}
StringBuilder sql = new StringBuilder(insertSql.substring(0, valuesIndex))
.append("VALUES (");
for (int i = 0; i < values.size(); i++) {
if (i > 0) {
sql.append(", ");
}
sql.append(literal(values.get(i)));
if (rows == null || rows.isEmpty()) {
throw new IllegalArgumentException("literal rows must not be empty");
}
return sql.append(')').toString();
StringBuilder sql = new StringBuilder(insertSql.substring(0, valuesIndex))
.append("VALUES ");
for (int rowIndex = 0; rowIndex < rows.size(); rowIndex++) {
if (rowIndex > 0) {
sql.append(' ');
}
List<Object> values = rows.get(rowIndex);
sql.append('(');
for (int i = 0; i < values.size(); i++) {
if (i > 0) {
sql.append(", ");
}
sql.append(literal(values.get(i)));
}
sql.append(')');
}
return sql.toString();
}
private static String literal(Object value) {

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

@@ -156,6 +156,25 @@ class TdengineJdbcHistoryWriterTest {
assertThat(jdbc.commits).isEqualTo(1);
}
@Test
void groupsLiteralRowsForSameChildTableIntoOneInsert() throws Exception {
RecordingJdbc jdbc = new RecordingJdbc();
TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema);
writer.appendRawFrames(List.of(
rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z")),
rawFrame("frame-2", Instant.parse("2026-06-29T05:00:02Z"))));
List<String> inserts = jdbc.executedSql.stream()
.filter(sql -> sql.startsWith("INSERT INTO raw_jt808_"))
.toList();
assertThat(inserts).hasSize(1);
assertThat(inserts.getFirst())
.contains("frame-1")
.contains("frame-2")
.contains(") (");
}
private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) {
return rawFrame(frameId, ts, "");
}