From 6016e3f99cf381abd5a46cf5b57ef3501727d30b Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 05:00:39 +0800 Subject: [PATCH] refactor: require explicit history consumer bindings --- ...icleHistoryKafkaConsumerConfiguration.java | 2 - .../historyapp/MavenModuleProfileTest.java | 14 ++++++ .../VehicleHistoryAppCompositionTest.java | 45 +++++++++++++++++++ 3 files changed, 59 insertions(+), 2 deletions(-) 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 53f53529..a46fa39f 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 @@ -27,11 +27,9 @@ public class VehicleHistoryKafkaConsumerConfiguration { SinkMqProperties props) { List workers = new KafkaEnvelopeConsumerFactory().createWorkers( Map.of( - "eventHistoryEnvelopeConsumerProcessor", processor, "eventHistoryGb32960EnvelopeConsumerProcessor", processor, "eventHistoryJt808EnvelopeConsumerProcessor", processor, "eventHistoryYutongMqttEnvelopeConsumerProcessor", processor, - "eventHistoryRawEnvelopeConsumerProcessor", rawProcessor, "eventHistoryGb32960RawEnvelopeConsumerProcessor", rawProcessor, "eventHistoryJt808RawEnvelopeConsumerProcessor", rawProcessor, "eventHistoryYutongMqttRawEnvelopeConsumerProcessor", rawProcessor), diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java index 3f7f48f8..c47a567a 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java @@ -29,6 +29,20 @@ class MavenModuleProfileTest { "modules/apps/command-gateway"); } + @Test + void xindaIsLegacyOnlyAndOutsideDefaultProductionReactor() throws Exception { + Document pom = rootPom(); + + assertThat(defaultModules(pom)) + .doesNotContain("modules/inbound/inbound-xinda-push") + .doesNotContain("modules/apps/xinda-push-app"); + + assertThat(profileModules(pom, "legacy-xinda")) + .containsExactly( + "modules/inbound/inbound-xinda-push", + "modules/apps/xinda-push-app"); + } + private static Document rootPom() throws Exception { DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true); 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 8b804d08..d5638d2f 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 @@ -1,6 +1,8 @@ package com.lingniu.ingest.historyapp; +import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor; import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink; +import com.lingniu.ingest.api.consumer.EnvelopeIngestResult; import com.lingniu.ingest.eventfilestore.EventFileStore; import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; @@ -18,6 +20,7 @@ import com.lingniu.ingest.sink.archive.RawArchiveEventSink; import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration; import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink; import com.lingniu.ingest.sink.mq.KafkaEventSink; +import com.lingniu.ingest.sink.mq.SinkMqProperties; import com.lingniu.ingest.sink.mq.proto.ParseStatusProto; import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload; @@ -39,6 +42,7 @@ import java.nio.file.Path; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; @@ -65,6 +69,23 @@ class VehicleHistoryAppCompositionTest { .doesNotContain("eventHistoryRawEnvelopeConsumerProcessor"); } + @Test + void kafkaConsumerRunnerRejectsGenericHistoryBindings() { + new ApplicationContextRunner() + .withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class) + .withBean("eventHistoryEnvelopeConsumerProcessor", EnvelopeConsumerProcessor.class, + () -> processor("event-history")) + .withBean("eventHistoryRawEnvelopeConsumerProcessor", EnvelopeConsumerProcessor.class, + () -> processor("event-history-raw")) + .withBean(SinkMqProperties.class, VehicleHistoryAppCompositionTest::genericOnlyHistoryConsumerProps) + .withPropertyValues("lingniu.ingest.sink.mq.consumer.enabled=true") + .run(context -> { + assertThat(context).hasFailed(); + assertThat(context.getStartupFailure()) + .hasRootCauseMessage("no vehicle history kafka consumer workers created; check consumer bindings"); + }); + } + @Test void createsTdengineHistoryStorageAndQueryBeansWithoutGb32960TcpServerOrEventFileStore() { new ApplicationContextRunner() @@ -160,6 +181,30 @@ class VehicleHistoryAppCompositionTest { return mock(KafkaProducer.class); } + private static EnvelopeConsumerProcessor processor(String service) { + return new EnvelopeConsumerProcessor( + service, + ignored -> EnvelopeIngestResult.processed("event-1", "VIN001"), + ignored -> {}); + } + + private static SinkMqProperties genericOnlyHistoryConsumerProps() { + SinkMqProperties props = new SinkMqProperties(); + props.setBootstrapServers("localhost:9092"); + props.getConsumer().setEnabled(true); + props.getConsumer().setBindings(Map.of( + "eventHistoryEnvelopeConsumerProcessor", binding("history-event", "vehicle.event.generic.v1"), + "eventHistoryRawEnvelopeConsumerProcessor", binding("history-raw", "vehicle.raw.generic.v1"))); + return props; + } + + private static SinkMqProperties.Binding binding(String groupId, String topic) { + SinkMqProperties.Binding binding = new SinkMqProperties.Binding(); + binding.setGroupId(groupId); + binding.setTopics(List.of(topic)); + return binding; + } + private static VehicleEnvelope rawFrameEnvelope() { return VehicleEnvelope.newBuilder() .setSchemaVersion("1.0")