From 662d04714d7bddf1e07162c233c97d085edeefcf Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 23 Jun 2026 15:35:51 +0800 Subject: [PATCH] docs: plan gb32960 service split --- .../plans/2026-06-23-gb32960-service-split.md | 1639 +++++++++++++++++ 1 file changed, 1639 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-23-gb32960-service-split.md diff --git a/docs/superpowers/plans/2026-06-23-gb32960-service-split.md b/docs/superpowers/plans/2026-06-23-gb32960-service-split.md new file mode 100644 index 00000000..0cbd3ae3 --- /dev/null +++ b/docs/superpowers/plans/2026-06-23-gb32960-service-split.md @@ -0,0 +1,1639 @@ +# GB32960 Service Split Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Split the current all-in-one GB32960 production path into separate ingest, history, and analytics runnable apps with Kafka as the durability boundary and ACK after Kafka success. + +**Architecture:** Add three Spring Boot app modules while keeping `bootstrap-all` available for development and rollback. First isolate runtime composition, then introduce GB32960-specific Kafka topics, then change the GB32960 ACK path so realtime/report ACKs are written after required Kafka production succeeds. History and analytics become Kafka consumers in separate apps and no longer share a JVM with TCP ingest. + +**Tech Stack:** Java 25, Spring Boot 3.5, Maven multi-module build, Netty, Kafka clients, Protobuf envelope, DuckDB/Parquet event store, local archive sink. + +--- + +## Current-State Notes + +- Existing all-in-one entrypoint: `modules/apps/bootstrap-all/src/main/java/com/lingniu/ingest/bootstrap/IngestApplication.java`. +- Existing all-in-one module: `modules/apps/bootstrap-all/pom.xml` imports protocols, sinks, history, state, stat, command gateway, MQTT, Xinda, and JT modules. +- Existing GB32960 handler: `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`. +- Existing handler currently writes report ACK before `dispatcher.dispatch(rf)`. +- Existing `DisruptorEventBus` publishes to sinks asynchronously and does not wait for sink completion. +- Existing `KafkaEventSink.accepts()` rejects `VehicleEvent.RawArchive`. +- Existing versioned topic names are not yet present; current topic names live under `lingniu.ingest.sink.mq.topics`. +- Existing history consumer logic: `modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java`. +- Existing analytics consumers: `VehicleStateEnvelopeIngestor` and `VehicleStatEnvelopeIngestor`. +- There is an unrelated working-tree change in `modules/apps/bootstrap-all/src/main/resources/application.yml` for GB32960 diagnostics capacity. Do not revert it. Do not include it in service-split commits unless the user explicitly asks. + +## File Structure + +Create app modules: + +- `modules/apps/gb32960-ingest-app/pom.xml` +- `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java` +- `modules/apps/gb32960-ingest-app/src/main/resources/application.yml` +- `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java` +- `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java` + +- `modules/apps/vehicle-history-app/pom.xml` +- `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java` +- `modules/apps/vehicle-history-app/src/main/resources/application.yml` +- `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java` +- `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java` + +- `modules/apps/vehicle-analytics-app/pom.xml` +- `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java` +- `modules/apps/vehicle-analytics-app/src/main/resources/application.yml` +- `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java` +- `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java` + +Modify shared build and Kafka contract files: + +- `pom.xml` +- `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java` +- `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/TopicRouter.java` +- `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java` +- `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java` +- `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java` + +Modify ACK/Kafka durability boundary files: + +- `modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/sink/EventSink.java` +- `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java` +- `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java` +- `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java` +- `modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java` + +Optional later follow-up files after the first split is running: + +- `docs/operations/gb32960-service-split-runbook.md` +- Deployment manifests or service manager files used by the user's environment. + +--- + +### Task 1: Register the Three New App Modules + +**Files:** +- Modify: `pom.xml` +- Create: `modules/apps/gb32960-ingest-app/pom.xml` +- Create: `modules/apps/vehicle-history-app/pom.xml` +- Create: `modules/apps/vehicle-analytics-app/pom.xml` + +- [ ] **Step 1: Add failing module-presence test by running Maven before modules exist** + +Run: + +```bash +export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home +export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH" +mvn -pl :gb32960-ingest-app -am validate +``` + +Expected: FAIL with a message that Maven cannot find project `:gb32960-ingest-app`. + +- [ ] **Step 2: Add modules to the root reactor** + +In `pom.xml`, add these module entries after `modules/apps/command-gateway` and before `modules/apps/bootstrap-all`: + +```xml + modules/apps/gb32960-ingest-app + modules/apps/vehicle-history-app + modules/apps/vehicle-analytics-app +``` + +Also add dependency management entries near the existing `command-gateway` entry: + +```xml + + com.lingniu.ingest + gb32960-ingest-app + ${project.version} + + + com.lingniu.ingest + vehicle-history-app + ${project.version} + + + com.lingniu.ingest + vehicle-analytics-app + ${project.version} + +``` + +- [ ] **Step 3: Create `gb32960-ingest-app` POM** + +Create `modules/apps/gb32960-ingest-app/pom.xml`: + +```xml + + + 4.0.0 + + com.lingniu.ingest + lingniu-vehicle-ingest + 0.1.0-SNAPSHOT + ../../../pom.xml + + + gb32960-ingest-app + gb32960-ingest-app + GB32960 protocol ingress runtime: TCP decode, auth, Kafka production, and ACK. + + + + com.lingniu.ingest + ingest-core + + + com.lingniu.ingest + session-core + + + com.lingniu.ingest + observability + + + com.lingniu.ingest + protocol-gb32960 + + + com.lingniu.ingest + sink-mq + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-test + test + + + + + gb32960-ingest-app + + + org.springframework.boot + spring-boot-maven-plugin + + --sun-misc-unsafe-memory-access=allow + + + + + repackage + + + + + + + +``` + +- [ ] **Step 4: Create `vehicle-history-app` POM** + +Create `modules/apps/vehicle-history-app/pom.xml`: + +```xml + + + 4.0.0 + + com.lingniu.ingest + lingniu-vehicle-ingest + 0.1.0-SNAPSHOT + ../../../pom.xml + + + vehicle-history-app + vehicle-history-app + Vehicle raw archive, event history, and GB32960 frame query runtime. + + + + com.lingniu.ingest + ingest-api + + + com.lingniu.ingest + observability + + + com.lingniu.ingest + sink-mq + + + com.lingniu.ingest + sink-archive + + + com.lingniu.ingest + event-file-store + + + com.lingniu.ingest + event-history-service + + + com.lingniu.ingest + protocol-gb32960 + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-test + test + + + + + vehicle-history-app + + + org.springframework.boot + spring-boot-maven-plugin + + --sun-misc-unsafe-memory-access=allow + + + + + repackage + + + + + + + +``` + +- [ ] **Step 5: Create `vehicle-analytics-app` POM** + +Create `modules/apps/vehicle-analytics-app/pom.xml`: + +```xml + + + 4.0.0 + + com.lingniu.ingest + lingniu-vehicle-ingest + 0.1.0-SNAPSHOT + ../../../pom.xml + + + vehicle-analytics-app + vehicle-analytics-app + Vehicle state, daily statistics, alarms, and analytics runtime. + + + + com.lingniu.ingest + ingest-api + + + com.lingniu.ingest + observability + + + com.lingniu.ingest + sink-mq + + + com.lingniu.ingest + vehicle-state-service + + + com.lingniu.ingest + vehicle-stat-service + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-test + test + + + + + vehicle-analytics-app + + + org.springframework.boot + spring-boot-maven-plugin + + --sun-misc-unsafe-memory-access=allow + + + + + repackage + + + + + + + +``` + +- [ ] **Step 6: Verify the modules are visible to Maven** + +Run: + +```bash +export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home +export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH" +mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am validate +``` + +Expected: BUILD SUCCESS. + +- [ ] **Step 7: Commit** + +```bash +git add pom.xml modules/apps/gb32960-ingest-app/pom.xml modules/apps/vehicle-history-app/pom.xml modules/apps/vehicle-analytics-app/pom.xml +git commit -m "build: add split service app modules" +``` + +--- + +### Task 2: Add App Entrypoints and Runtime Defaults + +**Files:** +- Create: `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java` +- Create: `modules/apps/gb32960-ingest-app/src/main/resources/application.yml` +- Create: `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java` +- Create: `modules/apps/vehicle-history-app/src/main/resources/application.yml` +- Create: `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java` +- Create: `modules/apps/vehicle-analytics-app/src/main/resources/application.yml` + +- [ ] **Step 1: Create the GB32960 ingest main class** + +Create `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`: + +```java +package com.lingniu.ingest.gb32960app; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication(scanBasePackages = "com.lingniu.ingest") +public class Gb32960IngestApplication { + + private static final Logger log = LoggerFactory.getLogger(Gb32960IngestApplication.class); + + public static void main(String[] args) { + try { + SpringApplication.run(Gb32960IngestApplication.class, args); + } catch (Throwable t) { + log.error("GB32960 ingest application failed to start, forcing JVM exit", t); + System.exit(1); + } + } +} +``` + +- [ ] **Step 2: Create the GB32960 ingest config** + +Create `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`: + +```yaml +spring: + application: + name: gb32960-ingest-app + threads: + virtual: + enabled: true + +server: + port: ${HTTP_PORT:20100} + +lingniu: + ingest: + gb32960: + enabled: true + port: ${GB32960_PORT:32960} + boss-threads: ${GB32960_BOSS_THREADS:1} + worker-threads: ${GB32960_WORKER_THREADS:0} + auth: + enabled: ${GB32960_AUTH_ENABLED:false} + case-sensitive: false + whitelist: [] + platforms: + - username: ${GB32960_PLATFORM_USER_HYUNDAI:Hyundai} + password: ${GB32960_PLATFORM_PWD_HYUNDAI:f2e3445d7cda409fb4f278f6fb890734} + allowed-ips: + - ${GB32960_PLATFORM_IP_HYUNDAI:8.134.95.166} + description: 外部下级平台 - Hyundai + tls: + enabled: ${GB32960_TLS_ENABLED:false} + cert-path: ${GB32960_TLS_CERT:} + key-path: ${GB32960_TLS_KEY:} + trust-cert-path: ${GB32960_TLS_CA:} + require-client-auth: true + vendor-extensions: + - name: guangdong-fc + match: + platform-accounts: + - Hyundai + vin-prefixes: [] + vins: [] + parse: + lenient-block-failure: true + diagnostics: + max-logged-frame-keys-per-channel: ${GB32960_DIAGNOSTICS_MAX_LOGGED_FRAME_KEYS_PER_CHANNEL:4096} + pipeline: + disruptor: + ring-buffer-size: ${PIPELINE_RING_BUFFER_SIZE:131072} + wait-strategy: ${PIPELINE_WAIT_STRATEGY:yielding} + producer-type: multi + dedup: + enabled: true + cache-size: 200000 + ttl-seconds: 600 + rate-limit: + per-vin-qps: 50 + session: + store: ${SESSION_STORE:memory} + ttl: ${SESSION_TTL:30m} + identity: + store: ${VEHICLE_IDENTITY_STORE:file} + file: + path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl} + sink: + mq: + enabled: ${KAFKA_ENABLED:true} + type: kafka + bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092} + compression-type: zstd + linger-ms: 20 + batch-size: 65536 + acks: all + enable-idempotence: true + node-id: ${KAFKA_NODE_ID:gb32960-ingest-local} + topics: + realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + location: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + alarm: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + session: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1} + raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1} + dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1} + consumer: + enabled: false + event-file-store: + enabled: false + event-history: + enabled: false + vehicle-state: + enabled: false + vehicle-stat: + enabled: false + +management: + endpoints: + web: + exposure: + include: health,info,metrics,prometheus,env + health: + redis: + enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false} + metrics: + tags: + application: gb32960-ingest-app + +logging: + level: + root: INFO + com.lingniu.ingest: INFO +``` + +- [ ] **Step 3: Create the history app main class** + +Create `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`: + +```java +package com.lingniu.ingest.historyapp; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication(scanBasePackages = "com.lingniu.ingest") +public class VehicleHistoryApplication { + + private static final Logger log = LoggerFactory.getLogger(VehicleHistoryApplication.class); + + public static void main(String[] args) { + try { + SpringApplication.run(VehicleHistoryApplication.class, args); + } catch (Throwable t) { + log.error("Vehicle history application failed to start, forcing JVM exit", t); + System.exit(1); + } + } +} +``` + +- [ ] **Step 4: Create the history app config** + +Create `modules/apps/vehicle-history-app/src/main/resources/application.yml`: + +```yaml +spring: + application: + name: vehicle-history-app + threads: + virtual: + enabled: true + +server: + port: ${HTTP_PORT:20200} + +lingniu: + ingest: + gb32960: + enabled: false + sink: + mq: + enabled: ${KAFKA_ENABLED:true} + type: kafka + bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092} + topics: + realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1} + dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1} + consumer: + enabled: ${KAFKA_CONSUMER_ENABLED:true} + client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-history} + auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest} + max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500} + bindings: + eventHistoryEnvelopeConsumerProcessor: + enabled: true + group-id: ${KAFKA_GROUP_HISTORY:vehicle-history} + topics: + - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + - ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1} + archive: + enabled: ${SINK_ARCHIVE_ENABLED:true} + type: local + path: ${SINK_ARCHIVE_PATH:./archive/} + event-file-store: + enabled: ${EVENT_FILE_STORE_ENABLED:true} + 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} + flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000} + event-history: + enabled: true + vehicle-state: + enabled: false + vehicle-stat: + enabled: false + +management: + endpoints: + web: + exposure: + include: health,info,metrics,prometheus,env + health: + redis: + enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false} + metrics: + tags: + application: vehicle-history-app +``` + +- [ ] **Step 5: Create the analytics app main class** + +Create `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`: + +```java +package com.lingniu.ingest.analyticsapp; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication(scanBasePackages = "com.lingniu.ingest") +public class VehicleAnalyticsApplication { + + private static final Logger log = LoggerFactory.getLogger(VehicleAnalyticsApplication.class); + + public static void main(String[] args) { + try { + SpringApplication.run(VehicleAnalyticsApplication.class, args); + } catch (Throwable t) { + log.error("Vehicle analytics application failed to start, forcing JVM exit", t); + System.exit(1); + } + } +} +``` + +- [ ] **Step 6: Create the analytics app config** + +Create `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`: + +```yaml +spring: + application: + name: vehicle-analytics-app + threads: + virtual: + enabled: true + +server: + port: ${HTTP_PORT:20300} + +lingniu: + ingest: + gb32960: + enabled: false + sink: + mq: + enabled: ${KAFKA_ENABLED:true} + type: kafka + bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092} + topics: + realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1} + consumer: + enabled: ${KAFKA_CONSUMER_ENABLED:true} + client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-analytics} + auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest} + max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500} + bindings: + vehicleStateEnvelopeConsumerProcessor: + enabled: ${VEHICLE_STATE_ENABLED:false} + group-id: ${KAFKA_GROUP_STATE:vehicle-state} + topics: + - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + vehicleStatEnvelopeConsumerProcessor: + enabled: ${VEHICLE_STAT_ENABLED:true} + group-id: ${KAFKA_GROUP_STAT:vehicle-stat} + topics: + - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + archive: + enabled: false + event-file-store: + enabled: false + event-history: + enabled: false + vehicle-state: + enabled: ${VEHICLE_STATE_ENABLED:false} + vehicle-stat: + enabled: ${VEHICLE_STAT_ENABLED:true} + file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/} + zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai} + +management: + endpoints: + web: + exposure: + include: health,info,metrics,prometheus,env + health: + redis: + enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false} + metrics: + tags: + application: vehicle-analytics-app +``` + +- [ ] **Step 7: Build all three app jars** + +Run: + +```bash +export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home +export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH" +mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true +``` + +Expected: BUILD SUCCESS and jars under each app module's `target/`. + +- [ ] **Step 8: Commit** + +```bash +git add modules/apps/gb32960-ingest-app modules/apps/vehicle-history-app modules/apps/vehicle-analytics-app +git commit -m "feat: add split service entrypoints" +``` + +--- + +### Task 3: Add Composition Tests for Service Boundaries + +**Files:** +- Create: `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java` +- Create: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java` +- Create: `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java` + +- [ ] **Step 1: Write GB32960 ingest composition test** + +Create `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`: + +```java +package com.lingniu.ingest.gb32960app; + +import com.lingniu.ingest.eventfilestore.EventFileStore; +import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration; +import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer; +import com.lingniu.ingest.sink.archive.ArchiveStore; +import com.lingniu.ingest.sink.mq.KafkaEventSink; +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +class Gb32960IngestAppCompositionTest { + + @Test + void gb32960IngestRuntimeStartsProtocolAndKafkaWithoutHistoryStorage() { + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(Gb32960AutoConfiguration.class)) + .withPropertyValues( + "lingniu.ingest.gb32960.enabled=true", + "lingniu.ingest.gb32960.port=0", + "lingniu.ingest.sink.mq.enabled=false", + "lingniu.ingest.sink.archive.enabled=false", + "lingniu.ingest.event-file-store.enabled=false", + "lingniu.ingest.event-history.enabled=false", + "lingniu.ingest.vehicle-state.enabled=false", + "lingniu.ingest.vehicle-stat.enabled=false") + .run(context -> { + assertThat(context).hasSingleBean(Gb32960NettyServer.class); + assertThat(context).doesNotHaveBean(ArchiveStore.class); + assertThat(context).doesNotHaveBean(EventFileStore.class); + assertThat(context).doesNotHaveBean(KafkaEventSink.class); + }); + } +} +``` + +If this test does not create required supporting beans, add the same auto-config classes used by existing bootstrap tests: `IngestCoreAutoConfiguration`, `SessionCoreAutoConfiguration`, and `VehicleIdentityAutoConfiguration`. + +- [ ] **Step 2: Run GB32960 ingest composition test** + +Run: + +```bash +mvn -pl :gb32960-ingest-app -Dtest=Gb32960IngestAppCompositionTest test +``` + +Expected after adding required auto-config classes: PASS. + +- [ ] **Step 3: Write history composition test** + +Create `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`: + +```java +package com.lingniu.ingest.historyapp; + +import com.lingniu.ingest.eventfilestore.EventFileStore; +import com.lingniu.ingest.eventhistory.EventHistoryController; +import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer; +import com.lingniu.ingest.sink.archive.ArchiveStore; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; + +class VehicleHistoryAppCompositionTest { + + @TempDir + Path tempDir; + + @Test + void historyRuntimeCreatesStorageAndQueryBeansWithoutProtocolListener() { + new ApplicationContextRunner() + .withPropertyValues( + "lingniu.ingest.gb32960.enabled=false", + "lingniu.ingest.sink.archive.enabled=true", + "lingniu.ingest.sink.archive.path=" + tempDir.resolve("archive"), + "lingniu.ingest.event-file-store.enabled=true", + "lingniu.ingest.event-file-store.path=" + tempDir.resolve("event-store"), + "lingniu.ingest.event-history.enabled=true", + "lingniu.ingest.vehicle-state.enabled=false", + "lingniu.ingest.vehicle-stat.enabled=false", + "lingniu.ingest.sink.mq.enabled=false") + .withConfiguration(AutoConfigurations.of( + com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration.class, + com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration.class, + com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration.class)) + .run(context -> { + assertThat(context).hasSingleBean(ArchiveStore.class); + assertThat(context).hasSingleBean(EventFileStore.class); + assertThat(context).hasSingleBean(EventHistoryController.class); + assertThat(context).doesNotHaveBean(Gb32960NettyServer.class); + }); + } +} +``` + +- [ ] **Step 4: Run history composition test** + +Run: + +```bash +mvn -pl :vehicle-history-app -Dtest=VehicleHistoryAppCompositionTest test +``` + +Expected: PASS. + +- [ ] **Step 5: Write analytics composition test** + +Create `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`: + +```java +package com.lingniu.ingest.analyticsapp; + +import com.lingniu.ingest.eventfilestore.EventFileStore; +import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer; +import com.lingniu.ingest.vehiclestat.DailyVehicleStatService; +import com.lingniu.ingest.vehiclestate.VehicleStateRepository; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; + +class VehicleAnalyticsAppCompositionTest { + + @TempDir + Path tempDir; + + @Test + void analyticsRuntimeCreatesStatisticsWithoutProtocolOrHistoryStorage() { + new ApplicationContextRunner() + .withPropertyValues( + "lingniu.ingest.gb32960.enabled=false", + "lingniu.ingest.sink.archive.enabled=false", + "lingniu.ingest.event-file-store.enabled=false", + "lingniu.ingest.event-history.enabled=false", + "lingniu.ingest.vehicle-state.enabled=false", + "lingniu.ingest.vehicle-stat.enabled=true", + "lingniu.ingest.vehicle-stat.file-path=" + tempDir.resolve("vehicle-stat"), + "lingniu.ingest.sink.mq.enabled=false") + .withConfiguration(AutoConfigurations.of( + com.lingniu.ingest.vehiclestat.config.VehicleStatAutoConfiguration.class, + com.lingniu.ingest.vehiclestate.config.VehicleStateAutoConfiguration.class)) + .run(context -> { + assertThat(context).hasSingleBean(DailyVehicleStatService.class); + assertThat(context).doesNotHaveBean(VehicleStateRepository.class); + assertThat(context).doesNotHaveBean(Gb32960NettyServer.class); + assertThat(context).doesNotHaveBean(EventFileStore.class); + }); + } +} +``` + +- [ ] **Step 6: Run analytics composition test** + +Run: + +```bash +mvn -pl :vehicle-analytics-app -Dtest=VehicleAnalyticsAppCompositionTest test +``` + +Expected: PASS. + +- [ ] **Step 7: Commit** + +```bash +git add modules/apps/gb32960-ingest-app/src/test modules/apps/vehicle-history-app/src/test modules/apps/vehicle-analytics-app/src/test +git commit -m "test: cover split service composition" +``` + +--- + +### Task 4: Version Kafka Topics for GB32960 Raw/Event/DLQ + +**Files:** +- Modify: `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java` +- Modify: `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/TopicRouter.java` +- Modify: `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java` +- Create or modify: `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java` +- Create or modify: `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java` + +- [ ] **Step 1: Write topic routing tests** + +Create `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java` if it does not exist. Add tests proving GB32960 raw and normalized events can route to versioned topics. Use the existing `VehicleEvent` constructors from `EventFileStoreSinkTest` and `EnvelopeMapperTelemetrySnapshotTest` as examples. + +Expected test names: + +```java +@Test +void rawArchiveRoutesToVersionedGb32960RawTopic() { + SinkMqProperties.Topics topics = new SinkMqProperties.Topics(); + topics.setRawArchive("vehicle.raw.gb32960.v1"); + TopicRouter router = new TopicRouter(topics); + + VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive( + "event-1", + "trace-1", + ProtocolId.GB32960, + "VIN001", + Instant.parse("2026-06-23T07:00:00Z"), + 0x02, + new byte[] {0x23, 0x23}, + Map.of("platformAccount", "Hyundai")); + + assertThat(router.route(raw)).isEqualTo("vehicle.raw.gb32960.v1"); +} +``` + +Also add a normalized realtime event test: + +```java +@Test +void realtimeRoutesToVersionedGb32960EventTopic() { + SinkMqProperties.Topics topics = new SinkMqProperties.Topics(); + topics.setRealtime("vehicle.event.gb32960.v1"); + TopicRouter router = new TopicRouter(topics); + + VehicleEvent.Realtime realtime = new VehicleEvent.Realtime( + "event-2", + "trace-2", + ProtocolId.GB32960, + "VIN001", + Instant.parse("2026-06-23T07:00:01Z"), + 0x02, + Map.of("soc", "80"), + Map.of("platformAccount", "Hyundai")); + + assertThat(router.route(realtime)).isEqualTo("vehicle.event.gb32960.v1"); +} +``` + +Adjust constructor arguments to match the current sealed `VehicleEvent` definitions in `modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java`. + +- [ ] **Step 2: Run topic router test and verify failure if topic defaults are still old** + +Run: + +```bash +mvn -pl :sink-mq -Dtest=TopicRouterTest test +``` + +Expected before implementation: compile failure if constructors are wrong or assertion failure if defaults are old. Fix constructors first; keep routing assertions. + +- [ ] **Step 3: Update `SinkMqProperties.Topics` defaults** + +In `SinkMqProperties.Topics`, change defaults: + +```java +private String realtime = "vehicle.event.gb32960.v1"; +private String location = "vehicle.event.gb32960.v1"; +private String alarm = "vehicle.event.gb32960.v1"; +private String session = "vehicle.event.gb32960.v1"; +private String mediaMeta = "vehicle.media.meta.v1"; +private String rawArchive = "vehicle.raw.gb32960.v1"; +private String dlq = "vehicle.dlq.gb32960.v1"; +``` + +- [ ] **Step 4: Allow raw archive records in KafkaEventSink** + +Change `KafkaEventSink.accepts()` to: + +```java +@Override +public boolean accepts(VehicleEvent event) { + return true; +} +``` + +Update the JavaDoc above it to say Kafka carries both normalized events and raw archive records in split-service mode. + +- [ ] **Step 5: Run sink-mq tests** + +Run: + +```bash +mvn -pl :sink-mq test +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java \ + modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/TopicRouter.java \ + modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java \ + modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java \ + modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java +git commit -m "feat: route gb32960 kafka topics" +``` + +--- + +### Task 5: Introduce an ACK-Aware GB32960 Dispatch Boundary + +**Files:** +- Modify: `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java` +- Modify: `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java` +- Modify: `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java` +- Create: `modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java` + +Design choice for first implementation: + +- Keep `Dispatcher.dispatch(RawFrame)` for existing async protocols and `bootstrap-all`. +- Add `Dispatcher.dispatchAndAwait(RawFrame, Duration)` returning `CompletableFuture` or throwing a checked runtime exception on timeout/failure. +- Implement `DisruptorEventBus.publishAndAwait(VehicleEvent, Duration)` by synchronously calling accepted sinks' `publish()` futures and waiting for all to complete. +- Use the synchronous path only from GB32960 report/heartbeat/session commands that require ACK after Kafka success. + +- [ ] **Step 1: Write failing ACK boundary test** + +Create `Gb32960ChannelHandlerAckBoundaryTest` using `EmbeddedChannel` and a fake dispatcher that records whether ACK happens before the future completes. + +Core assertion: + +```java +@Test +void realtimeReportAckWaitsForKafkaDurabilityBoundary() { + CompletableFuture durability = new CompletableFuture<>(); + RecordingAckService ackService = new RecordingAckService(); + AwaitingDispatcher dispatcher = new AwaitingDispatcher(durability); + Gb32960ChannelHandler handler = handlerWith(dispatcher, ackService); + + EmbeddedChannel channel = new EmbeddedChannel(handler); + channel.writeInbound(validRealtimeFrameBytes()); + + assertThat(ackService.writtenTags()).isEmpty(); + + durability.complete(null); + channel.runPendingTasks(); + + assertThat(ackService.writtenTags()).contains("report-ack-0x2"); +} +``` + +Implementation notes: + +- Reuse valid frame fixture builders from existing GB32960 codec tests. +- If `Gb32960AckService` is hard to spy because methods are concrete, subclass it in test and override `writeResponse`/`writeAndClose` to record tags instead of writing bytes. +- If `Dispatcher` is not subclassable, extract a small interface such as `FrameDispatcher` in `ingest-core` and adapt `Dispatcher` to implement it. + +- [ ] **Step 2: Run the failing test** + +Run: + +```bash +mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test +``` + +Expected before implementation: FAIL because current handler writes ACK before dispatch completion. + +- [ ] **Step 3: Add await API to event bus** + +In `DisruptorEventBus`, add: + +```java +public CompletableFuture publishAndAwait(VehicleEvent event) { + List> futures = sinks.stream() + .filter(sink -> sink.accepts(event)) + .map(sink -> sink.publish(event)) + .toList(); + if (futures.isEmpty()) { + return CompletableFuture.completedFuture(null); + } + return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)); +} +``` + +This requires promoting `sinks` from constructor-local variable to a field: + +```java +private final List sinks; +``` + +Set it in the constructor: + +```java +this.sinks = List.copyOf(sinks); +``` + +- [ ] **Step 4: Add await API to Dispatcher** + +In `Dispatcher`, add: + +```java +public CompletableFuture dispatchAndAwait(RawFrame frame) { + List events = mapFrameToEvents(frame); + CompletableFuture[] futures = events.stream() + .map(eventBus::publishAndAwait) + .toArray(CompletableFuture[]::new); + return CompletableFuture.allOf(futures); +} +``` + +Refactor existing `dispatch(RawFrame)` to reuse `mapFrameToEvents(frame)`: + +```java +public void dispatch(RawFrame frame) { + for (VehicleEvent e : mapFrameToEvents(frame)) { + eventBus.publish(e); + } +} +``` + +If `Dispatcher` currently interleaves raw archive event publication and handler mapping, move that logic into a private method: + +```java +private List mapFrameToEvents(RawFrame frame) { + List events = new ArrayList<>(); + events.add(toRawArchive(frame)); + events.addAll(handlerEvents(frame)); + return events; +} +``` + +Use exact existing helper names from `Dispatcher.java`; do not duplicate mapping logic. + +- [ ] **Step 5: Move GB32960 ACK after dispatch completion** + +In `Gb32960ChannelHandler.channelRead0`, for commands that currently ACK before `dispatcher.dispatch(rf)`, build `RawFrame` first, call `dispatcher.dispatchAndAwait(rf)`, and write ACK in the completion callback: + +```java +dispatcher.dispatchAndAwait(rf).whenComplete((ignored, ex) -> { + if (ex != null) { + log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}", + addr(ctx), vin, Integer.toHexString(cmd.code()), ex); + ctx.close(); + return; + } + writeAckForCommand(ctx, msg, rawVin); +}); +``` + +Extract current ACK logic into command-specific method: + +```java +private void writeAckForCommand(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + switch (msg.header().command()) { + case VEHICLE_LOGIN -> writeVehicleLoginAck(ctx, msg, rawVin); + case VEHICLE_LOGOUT -> writeVehicleLogoutAck(ctx, msg, rawVin); + case PLATFORM_LOGIN -> writePlatformLoginAck(ctx, msg, rawVin); + case PLATFORM_LOGOUT -> writePlatformLogoutAck(ctx, msg, rawVin); + case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> writeReportOrHeartbeatAck(ctx, msg, rawVin); + case TIME_CALIBRATION -> writeTimeCalibrationAck(ctx, msg, rawVin); + default -> { } + } +} +``` + +Keep authorization failure ACKs immediate because they are not accepted frames and do not cross the Kafka durability boundary. + +- [ ] **Step 6: Run ACK boundary test** + +Run: + +```bash +mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test +``` + +Expected: PASS. + +- [ ] **Step 7: Run core dispatcher tests** + +Run: + +```bash +mvn -pl :ingest-core test +``` + +Expected: PASS. + +- [ ] **Step 8: Commit** + +```bash +git add modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java \ + modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java \ + modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java \ + modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java +git commit -m "feat: ack gb32960 after kafka boundary" +``` + +--- + +### Task 6: Wire History and Analytics Consumer Bindings + +**Files:** +- Modify: `modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java` +- Modify: `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java` +- Modify: `modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java` +- Modify tests under the same service modules. + +- [ ] **Step 1: Verify event-history accepts raw and normalized envelopes** + +Run: + +```bash +rg -n "RawArchive|raw_archive|TelemetrySnapshot|Envelope" modules/services/event-history-service/src/main/java modules/services/event-history-service/src/test/java +``` + +Expected: identify whether `EventHistoryEnvelopeIngestor` already persists raw archive envelopes. If raw archive envelopes are ignored, continue with Step 2. + +- [ ] **Step 2: Add event-history raw archive ingestion test** + +Add a test to the existing event-history test suite that creates an envelope with `raw_archive` populated and asserts an `EventFileRecord` with `eventType = RAW_ARCHIVE` is written with `rawArchiveUri`. + +Expected assertion shape: + +```java +assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE"); +assertThat(record.rawArchiveUri()).isEqualTo("archive://2026/06/23/GB32960/VIN001/event-1.bin"); +``` + +- [ ] **Step 3: Implement raw archive ingestion if missing** + +In `EventHistoryEnvelopeIngestor`, map raw archive envelopes to `EventFileRecord` using the same conventions as `EventFileStoreSink`. + +Implementation rule: + +- Do not embed raw bytes into Parquet history records. +- Store `rawArchiveUri`, event id, protocol, VIN, command, event time, and metadata. + +- [ ] **Step 4: Verify analytics consumers ignore raw topic records** + +Add or update tests for `VehicleStatEnvelopeIngestor` and `VehicleStateEnvelopeIngestor`: + +```java +@Test +void ignoresRawArchiveEnvelopeForAnalytics() { + ConsumerProcessResult result = ingestor.ingest(rawArchiveEnvelopeRecord()); + assertThat(result.success()).isTrue(); + assertThat(repositoryWrites()).isEmpty(); +} +``` + +Use the actual result type currently returned by the ingestor. + +- [ ] **Step 5: Run service consumer tests** + +Run: + +```bash +mvn -pl :event-history-service,:vehicle-state-service,:vehicle-stat-service test +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add modules/services/event-history-service modules/services/vehicle-state-service modules/services/vehicle-stat-service +git commit -m "feat: split history and analytics consumers" +``` + +--- + +### Task 7: Add App Default Tests + +**Files:** +- Create: `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java` +- Create: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java` +- Create: `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java` + +- [ ] **Step 1: Add YAML property loader helper to each test** + +Use this helper in each defaults test: + +```java +private static Properties loadApplicationProperties() { + YamlPropertiesFactoryBean yaml = new YamlPropertiesFactoryBean(); + yaml.setResources(new ClassPathResource("application.yml")); + Properties props = yaml.getObject(); + assertThat(props).isNotNull(); + return props; +} +``` + +Imports: + +```java +import org.springframework.beans.factory.config.YamlPropertiesFactoryBean; +import org.springframework.core.io.ClassPathResource; +import java.util.Properties; +import static org.assertj.core.api.Assertions.assertThat; +``` + +- [ ] **Step 2: Assert GB32960 ingest defaults** + +Create `Gb32960IngestAppDefaultsTest`: + +```java +@Test +void gb32960IngestDefaultsOnlyEnableProtocolAndKafkaProducer() { + Properties props = loadApplicationProperties(); + + assertThat(props.getProperty("spring.application.name")).isEqualTo("gb32960-ingest-app"); + assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("true"); + assertThat(props.getProperty("lingniu.ingest.gb32960.port")).isEqualTo("${GB32960_PORT:32960}"); + assertThat(props.getProperty("lingniu.ingest.sink.mq.enabled")).isEqualTo("${KAFKA_ENABLED:true}"); + assertThat(props.getProperty("lingniu.ingest.sink.mq.consumer.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.vehicle-state.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("false"); +} +``` + +- [ ] **Step 3: Assert history defaults** + +Create `VehicleHistoryAppDefaultsTest`: + +```java +@Test +void historyDefaultsEnableStorageAndHistoryConsumerOnly() { + Properties props = loadApplicationProperties(); + + assertThat(props.getProperty("spring.application.name")).isEqualTo("vehicle-history-app"); + assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.sink.archive.enabled")).isEqualTo("${SINK_ARCHIVE_ENABLED:true}"); + assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("${EVENT_FILE_STORE_ENABLED:true}"); + assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("true"); + assertThat(props.getProperty("lingniu.ingest.vehicle-state.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.group-id")) + .isEqualTo("${KAFKA_GROUP_HISTORY:vehicle-history}"); +} +``` + +- [ ] **Step 4: Assert analytics defaults** + +Create `VehicleAnalyticsAppDefaultsTest`: + +```java +@Test +void analyticsDefaultsEnableStatConsumerAndDisableProtocolAndHistoryStorage() { + Properties props = loadApplicationProperties(); + + assertThat(props.getProperty("spring.application.name")).isEqualTo("vehicle-analytics-app"); + assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.sink.archive.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("false"); + assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("${VEHICLE_STAT_ENABLED:true}"); + assertThat(props.getProperty("lingniu.ingest.sink.mq.consumer.bindings.vehicleStatEnvelopeConsumerProcessor.group-id")) + .isEqualTo("${KAFKA_GROUP_STAT:vehicle-stat}"); +} +``` + +- [ ] **Step 5: Run app default tests** + +Run: + +```bash +mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java \ + modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java \ + modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java +git commit -m "test: assert split app defaults" +``` + +--- + +### Task 8: End-to-End Verification With Local Kafka + +**Files:** +- Create: `docs/operations/gb32960-service-split-runbook.md` + +- [ ] **Step 1: Build all modules** + +Run: + +```bash +export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home +export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH" +mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true +``` + +Expected: BUILD SUCCESS. + +- [ ] **Step 2: Start Kafka test environment** + +Use the repository's existing Kafka setup if present. If there is no repository-local Kafka script, use a local Kafka compatible with `KAFKA_BROKERS=127.0.0.1:9092`. + +Create topics: + +```bash +kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.raw.gb32960.v1 --partitions 12 --replication-factor 1 +kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.event.gb32960.v1 --partitions 12 --replication-factor 1 +kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.dlq.gb32960.v1 --partitions 3 --replication-factor 1 +``` + +- [ ] **Step 3: Start split services on local ports** + +Run each command in a separate terminal: + +```bash +KAFKA_BROKERS=127.0.0.1:9092 \ +GB32960_PORT=32960 \ +HTTP_PORT=20100 \ +java --sun-misc-unsafe-memory-access=allow \ + -jar modules/apps/gb32960-ingest-app/target/gb32960-ingest-app.jar +``` + +```bash +KAFKA_BROKERS=127.0.0.1:9092 \ +HTTP_PORT=20200 \ +EVENT_FILE_STORE_PATH=./target/split-event-store \ +SINK_ARCHIVE_PATH=./target/split-archive \ +java --sun-misc-unsafe-memory-access=allow \ + -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar +``` + +```bash +KAFKA_BROKERS=127.0.0.1:9092 \ +HTTP_PORT=20300 \ +VEHICLE_STAT_FILE_PATH=./target/split-vehicle-stat \ +java --sun-misc-unsafe-memory-access=allow \ + -jar modules/apps/vehicle-analytics-app/target/vehicle-analytics-app.jar +``` + +- [ ] **Step 4: Verify health endpoints** + +Run: + +```bash +curl -sS http://127.0.0.1:20100/actuator/health +curl -sS http://127.0.0.1:20200/actuator/health +curl -sS http://127.0.0.1:20300/actuator/health +``` + +Expected: each returns `{"status":"UP"}`. + +- [ ] **Step 5: Send a known-valid GB32960 frame** + +Use an existing GB32960 test fixture from `modules/protocols/protocol-gb32960/src/test/java`. If no command-line sender exists, add a small test-only utility or use `nc` with a hex-decoded fixture: + +```bash +xxd -r -p /tmp/gb32960-valid-realtime.hex | nc 127.0.0.1 32960 +``` + +Expected: + +- Kafka topic `vehicle.raw.gb32960.v1` receives one raw record. +- Kafka topic `vehicle.event.gb32960.v1` receives one or more normalized records. +- Client receives a GB32960 success ACK. +- `target/split-archive` contains a `.bin` file. +- `target/split-event-store` contains `events.duckdb` or Parquet files after flush. +- `target/split-vehicle-stat` contains updated stat files after analytics consumes the event. + +- [ ] **Step 6: Write runbook** + +Create `docs/operations/gb32960-service-split-runbook.md` with: + +```markdown +# GB32960 Split Service Runbook + +## Services + +- `gb32960-ingest-app`: TCP 32960, HTTP 20100 +- `vehicle-history-app`: HTTP 20200 +- `vehicle-analytics-app`: HTTP 20300 + +## Kafka Topics + +- `vehicle.raw.gb32960.v1` +- `vehicle.event.gb32960.v1` +- `vehicle.dlq.gb32960.v1` + +## ACK Semantics + +GB32960 success ACK is sent after required Kafka production succeeds. History and analytics failures do not block ACK. + +## Local Start + +Use the commands from Task 8 Step 3. + +## Verification + +Use health endpoints, Kafka topic consumption, archive file count, event-store query APIs, analytics output files, and ACK latency metrics. + +## Rollback + +Stop split services and start `bootstrap-all` with the previous environment. Keep `bootstrap-all` available until split service counts and sample queries match production expectations. +``` + +- [ ] **Step 7: Commit** + +```bash +git add docs/operations/gb32960-service-split-runbook.md +git commit -m "docs: add gb32960 split service runbook" +``` + +--- + +### Task 9: Final Verification and Cutover Checklist + +**Files:** +- Modify: `docs/operations/gb32960-service-split-runbook.md` + +- [ ] **Step 1: Run targeted tests** + +Run: + +```bash +mvn -pl :sink-mq,:ingest-core,:protocol-gb32960,:event-history-service,:vehicle-state-service,:vehicle-stat-service test +``` + +Expected: BUILD SUCCESS. + +- [ ] **Step 2: Run split app tests** + +Run: + +```bash +mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test +``` + +Expected: BUILD SUCCESS. + +- [ ] **Step 3: Run package verification** + +Run: + +```bash +mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true +``` + +Expected: BUILD SUCCESS. + +- [ ] **Step 4: Update runbook with measured results** + +Append a section: + +```markdown +## Latest Local Verification + +- Build: +- Unit tests: +- Split app startup: +- Kafka raw records: +- Kafka event records: +- Archive files: +- Event-store files: +- Analytics output: +- ACK behavior: +``` + +Fill the fields with the actual command outputs and counts from the run. + +- [ ] **Step 5: Commit final verification docs** + +```bash +git add docs/operations/gb32960-service-split-runbook.md +git commit -m "docs: record gb32960 split verification" +``` + +--- + +## Self-Review + +Spec coverage: + +- Three services are covered by Tasks 1, 2, 3, and 7. +- Kafka topic and routing contract is covered by Task 4. +- ACK after Kafka success is covered by Task 5. +- History consumption and storage is covered by Task 6 and Task 8. +- Analytics consumption and processing is covered by Task 6 and Task 8. +- `bootstrap-all` remains untouched except for pre-existing unrelated worktree changes. +- Verification and runbook are covered by Tasks 8 and 9. + +Implementation risk notes: + +- Task 5 is the most invasive task because it changes protocol ACK ordering. Implement it behind the smallest possible new API and keep `Dispatcher.dispatch(RawFrame)` behavior for other protocols. +- If Kafka transactions are needed to atomically produce raw and normalized records, add that as a follow-up after the split app boundary is working. The first implementation can use idempotent producer plus shared `eventId`/`sourceRawEventId`. +- If composition tests need additional auto-config classes, copy the exact pattern from `ProtocolAutoConfigurationCompositionTest` rather than widening app dependencies.