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

feat: productionize raw history ingestion

Browse Source
This commit is contained in:
lingniu
2026-06-30 23:21:58 +08:00
parent 3cc7ac9669
commit cbba617801
100 changed files with 2995 additions and 1697 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/EnvelopeMapper.java
View File

@@ -92,6 +92,7 @@ public final class EnvelopeMapper {
.setUri(uri)
.setChecksum(checksum)
.setSizeBytes(size)
.setParsedJson(nullToEmpty(ra.parsedJson()))
.build());
b.setRawFrameFact(com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload.newBuilder()
.setFrameId(frameId)

48
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerRunner.java
View File

@@ -10,17 +10,19 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
public final class KafkaEnvelopeConsumerRunner implements SmartLifecycle, AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(KafkaEnvelopeConsumerRunner.class);
private final List<KafkaEnvelopeConsumerWorker> workers;
private final Supplier<List<KafkaEnvelopeConsumerWorker>> workersSupplier;
private final Duration pollTimeout;
private final Duration loopBackoff;
private final boolean autoStartup;
private final AtomicBoolean running = new AtomicBoolean(false);
private final AtomicBoolean closed = new AtomicBoolean(false);
private volatile List<KafkaEnvelopeConsumerWorker> workers;
private ExecutorService executor;
public KafkaEnvelopeConsumerRunner(List<KafkaEnvelopeConsumerWorker> workers,
@@ -30,14 +32,28 @@ public final class KafkaEnvelopeConsumerRunner implements SmartLifecycle, AutoCl
if (workers == null || workers.isEmpty()) {
throw new IllegalArgumentException("workers must not be empty");
}
this.workersSupplier = () -> List.copyOf(workers);
this.workers = List.copyOf(workers);
this.pollTimeout = pollTimeout == null ? Duration.ofSeconds(1) : pollTimeout;
this.loopBackoff = loopBackoff == null ? Duration.ofSeconds(1) : loopBackoff;
this.autoStartup = autoStartup;
}
public KafkaEnvelopeConsumerRunner(Supplier<List<KafkaEnvelopeConsumerWorker>> workersSupplier,
Duration pollTimeout,
Duration loopBackoff,
boolean autoStartup) {
if (workersSupplier == null) {
throw new IllegalArgumentException("workersSupplier must not be null");
}
this.workersSupplier = workersSupplier;
this.pollTimeout = pollTimeout == null ? Duration.ofSeconds(1) : pollTimeout;
this.loopBackoff = loopBackoff == null ? Duration.ofSeconds(1) : loopBackoff;
this.autoStartup = autoStartup;
}
public List<KafkaEnvelopeConsumerWorker> workers() {
return workers;
return workersOrCreate();
}
@Override
@@ -45,17 +61,35 @@ public final class KafkaEnvelopeConsumerRunner implements SmartLifecycle, AutoCl
if (!running.compareAndSet(false, true)) {
return;
}
List<KafkaEnvelopeConsumerWorker> activeWorkers = workersOrCreate();
// 每个 worker 一条后台线程，避免某个处理器阻塞时拖慢其他消费组。
executor = Executors.newFixedThreadPool(workers.size(), r -> {
executor = Executors.newFixedThreadPool(activeWorkers.size(), r -> {
Thread thread = new Thread(r, "kafka-envelope-consumer");
thread.setDaemon(true);
return thread;
});
for (KafkaEnvelopeConsumerWorker worker : workers) {
for (KafkaEnvelopeConsumerWorker worker : activeWorkers) {
executor.submit(() -> pollLoop(worker));
}
}
private List<KafkaEnvelopeConsumerWorker> workersOrCreate() {
List<KafkaEnvelopeConsumerWorker> current = workers;
if (current != null) {
return current;
}
synchronized (this) {
if (workers == null) {
List<KafkaEnvelopeConsumerWorker> created = workersSupplier.get();
if (created == null || created.isEmpty()) {
throw new IllegalStateException("no kafka envelope consumer workers created; check consumer bindings");
}
workers = List.copyOf(created);
}
return workers;
}
}
private void pollLoop(KafkaEnvelopeConsumerWorker worker) {
while (running.get()) {
try {
@@ -121,7 +155,11 @@ public final class KafkaEnvelopeConsumerRunner implements SmartLifecycle, AutoCl
if (!closed.compareAndSet(false, true)) {
return;
}
for (KafkaEnvelopeConsumerWorker worker : workers) {
List<KafkaEnvelopeConsumerWorker> current = workers;
if (current == null) {
return;
}
for (KafkaEnvelopeConsumerWorker worker : current) {
worker.close();
}
}

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

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

21
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqAutoConfiguration.java
View File

@@ -1,6 +1,5 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
@@ -8,15 +7,12 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;
/**
@@ -100,21 +96,4 @@ public class SinkMqAutoConfiguration {
return new KafkaEnvelopeDeadLetterSink(producer, props.getTopics().getDlq());
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnBean(EnvelopeConsumerProcessor.class)
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq.consumer", name = "enabled", havingValue = "true")
public KafkaEnvelopeConsumerRunner kafkaEnvelopeConsumerRunner(Map<String, EnvelopeConsumerProcessor> processors,
SinkMqProperties props) {
// Consumer runner 根据 processor bean 名和配置 binding 生成 worker；没有 binding 时直接失败，避免静默不消费。
List<KafkaEnvelopeConsumerWorker> workers = new KafkaEnvelopeConsumerFactory().createWorkers(processors, props);
if (workers.isEmpty()) {
throw new IllegalStateException("no kafka envelope consumer workers created; check consumer bindings");
}
return new KafkaEnvelopeConsumerRunner(
workers,
Duration.ofMillis(props.getConsumer().getPollTimeoutMillis()),
Duration.ofMillis(props.getConsumer().getLoopBackoffMillis()),
props.getConsumer().isAutoStartup());
}
}

42
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqConsumerAutoConfiguration.java Normal file
View File

@@ -0,0 +1,42 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import java.time.Duration;
import java.util.List;
import java.util.Map;
@AutoConfiguration(after = SinkMqAutoConfiguration.class)
@AutoConfigureAfter(name = {
"com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration",
"com.lingniu.ingest.vehiclestate.config.VehicleStateAutoConfiguration",
"com.lingniu.ingest.vehiclestat.config.VehicleStatAutoConfiguration"
})
@EnableConfigurationProperties(SinkMqProperties.class)
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq", name = "enabled", havingValue = "true", matchIfMissing = true)
public class SinkMqConsumerAutoConfiguration {
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq.consumer", name = "enabled", havingValue = "true")
public KafkaEnvelopeConsumerRunner kafkaEnvelopeConsumerRunner(ListableBeanFactory beanFactory,
SinkMqProperties props) {
return new KafkaEnvelopeConsumerRunner(
() -> createWorkers(beanFactory, props),
Duration.ofMillis(props.getConsumer().getPollTimeoutMillis()),
Duration.ofMillis(props.getConsumer().getLoopBackoffMillis()),
props.getConsumer().isAutoStartup());
}
private List<KafkaEnvelopeConsumerWorker> createWorkers(ListableBeanFactory beanFactory, SinkMqProperties props) {
Map<String, EnvelopeConsumerProcessor> processors = beanFactory.getBeansOfType(EnvelopeConsumerProcessor.class);
return new KafkaEnvelopeConsumerFactory().createWorkers(processors, props);
}
}

1
modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto
View File

@@ -45,6 +45,7 @@ message RawArchiveRef {
string uri = 1;
string checksum = 2;
int64 size_bytes = 3;
string parsed_json = 4;
}
message TelemetrySnapshot {

1
modules/sinks/sink-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@@ -1 +1,2 @@
com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration
com.lingniu.ingest.sink.mq.SinkMqConsumerAutoConfiguration

37
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEnvelopeConsumerWorkerTest.java
View File

@@ -1,6 +1,7 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerRecord;
import com.lingniu.ingest.api.consumer.EnvelopeBatchIngestor;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterRecord;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
@@ -14,6 +15,7 @@ import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
@@ -44,4 +46,39 @@ class KafkaEnvelopeConsumerWorkerTest {
assertThat(record.payload()).containsExactly(0x01, 0x02);
});
}
@Test
void pollsKafkaRecordsAndDispatchesBatchWhenProcessorSupportsIt() {
MockConsumer<String, byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
TopicPartition partition = new TopicPartition("vehicle.raw", 0);
consumer.assign(List.of(partition));
consumer.updateBeginningOffsets(Map.of(partition, 0L));
consumer.addRecord(new ConsumerRecord<>("vehicle.raw", 0, 12L, "VIN001", new byte[]{0x01}));
consumer.addRecord(new ConsumerRecord<>("vehicle.raw", 0, 13L, "VIN002", new byte[]{0x02}));
AtomicInteger batchCalls = new AtomicInteger();
EnvelopeConsumerProcessor processor = new EnvelopeConsumerProcessor(
"event-history",
new EnvelopeBatchIngestor() {
@Override
public EnvelopeIngestResult tryIngest(byte[] kafkaValue) {
throw new AssertionError("single-record ingest should not be used for a batch-capable processor");
}
@Override
public List<EnvelopeIngestResult> tryIngestAll(List<byte[]> kafkaValues) {
batchCalls.incrementAndGet();
assertThat(kafkaValues).hasSize(2);
return List.of(
EnvelopeIngestResult.stored("event-1", "VIN001"),
EnvelopeIngestResult.invalid("bad envelope"));
}
},
record -> {});
int processed = new KafkaEnvelopeConsumerWorker(
consumer, Map.of("vehicle.raw", processor)).pollOnce(Duration.ZERO);
assertThat(processed).isEqualTo(2);
assertThat(batchCalls).hasValue(1);
}
}

2
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/SinkMqConsumerAutoConfigurationTest.java
View File

@@ -10,7 +10,7 @@ import static org.assertj.core.api.Assertions.assertThat;
class SinkMqConsumerAutoConfigurationTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(SinkMqAutoConfiguration.class)
.withUserConfiguration(SinkMqAutoConfiguration.class, SinkMqConsumerAutoConfiguration.class)
.withBean("vehicleStateEnvelopeConsumerProcessor", EnvelopeConsumerProcessor.class,
() -> new EnvelopeConsumerProcessor(
"vehicle-state",