shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
662d04714d7bddf1e07162c233c97d085edeefcf
lingniu-vehicle-ingest/docs/superpowers/plans/2026-06-23-gb32960-service-split.md
lingniu 662d04714d docs: plan gb32960 service split
2026-06-23 15:35:51 +08:00

1640 lines
58 KiB
Markdown
Raw Blame History

# 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
<module>modules/apps/gb32960-ingest-app</module>
<module>modules/apps/vehicle-history-app</module>
<module>modules/apps/vehicle-analytics-app</module>
```
Also add dependency management entries near the existing `command-gateway` entry:
```xml
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>gb32960-ingest-app</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>vehicle-history-app</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>vehicle-analytics-app</artifactId>
<version>${project.version}</version>
</dependency>
```
- [ ] **Step 3: Create `gb32960-ingest-app` POM**
Create `modules/apps/gb32960-ingest-app/pom.xml`:
```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lingniu.ingest</groupId>
<artifactId>lingniu-vehicle-ingest</artifactId>
<version>0.1.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<artifactId>gb32960-ingest-app</artifactId>
<name>gb32960-ingest-app</name>
<description>GB32960 protocol ingress runtime: TCP decode, auth, Kafka production, and ACK.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-core</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>session-core</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>observability</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>protocol-gb32960</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-mq</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>gb32960-ingest-app</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
</configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
```
- [ ] **Step 4: Create `vehicle-history-app` POM**
Create `modules/apps/vehicle-history-app/pom.xml`:
```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lingniu.ingest</groupId>
<artifactId>lingniu-vehicle-ingest</artifactId>
<version>0.1.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<artifactId>vehicle-history-app</artifactId>
<name>vehicle-history-app</name>
<description>Vehicle raw archive, event history, and GB32960 frame query runtime.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-api</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>observability</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-mq</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-archive</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>event-file-store</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>event-history-service</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>protocol-gb32960</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>vehicle-history-app</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
</configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
```
- [ ] **Step 5: Create `vehicle-analytics-app` POM**
Create `modules/apps/vehicle-analytics-app/pom.xml`:
```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lingniu.ingest</groupId>
<artifactId>lingniu-vehicle-ingest</artifactId>
<version>0.1.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<artifactId>vehicle-analytics-app</artifactId>
<name>vehicle-analytics-app</name>
<description>Vehicle state, daily statistics, alarms, and analytics runtime.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-api</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>observability</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-mq</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>vehicle-state-service</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>vehicle-stat-service</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>vehicle-analytics-app</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
</configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
```
- [ ] **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<Void>` 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<Void> 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<Void> publishAndAwait(VehicleEvent event) {
List<CompletableFuture<Void>> 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<EventSink> 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<Void> dispatchAndAwait(RawFrame frame) {
List<VehicleEvent> 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<VehicleEvent> mapFrameToEvents(RawFrame frame) {
List<VehicleEvent> 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.
Reference in New Issue View Git Blame Copy Permalink