shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
80c051e056e15568c00599a814501bc26fdb88a9
lingniu-vehicle-ingest/docs/superpowers/plans/2026-06-23-gb32960-service-split.md
lingniu e80fa6f38c refactor: rename kafka sink module
2026-07-01 10:56:33 +08:00

59 KiB
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.kafka.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-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java
  • modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/TopicRouter.java
  • modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaEventSink.java
  • modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java
  • modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/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:

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:

        <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:

            <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 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-kafka</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 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-kafka</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 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-kafka</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:

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
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:

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:

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:redis}
      ttl: ${SESSION_TTL:30m}
    identity:
      store: ${VEHICLE_IDENTITY_STORE:mysql}
      mysql:
        jdbc-url: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:jdbc:mysql://127.0.0.1:3306/lingniu_vehicle?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai}
        username: ${VEHICLE_IDENTITY_MYSQL_USERNAME:root}
        password: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:}
        table-name: ${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_bindings}
        initialize-schema: ${VEHICLE_IDENTITY_MYSQL_INITIALIZE_SCHEMA:true}
    sink:
      kafka:
        enabled: ${KAFKA_ENABLED:true}
        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:

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:

spring:
  application:
    name: vehicle-history-app
  threads:
    virtual:
      enabled: true

server:
  port: ${HTTP_PORT:20200}

lingniu:
  ingest:
    gb32960:
      enabled: false
    sink:
      kafka:
        enabled: ${KAFKA_ENABLED:true}
        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:

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:

spring:
  application:
    name: vehicle-analytics-app
  threads:
    virtual:
      enabled: true

server:
  port: ${HTTP_PORT:20300}

lingniu:
  ingest:
    gb32960:
      enabled: false
    sink:
      kafka:
        enabled: ${KAFKA_ENABLED:true}
        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:

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
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:

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.kafka.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.kafka.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:

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:

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.kafka.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:

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:

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.kafka.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:

mvn -pl :vehicle-analytics-app -Dtest=VehicleAnalyticsAppCompositionTest test

Expected: PASS.

  • Step 7: Commit
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-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java

  • Modify: modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/TopicRouter.java

  • Modify: modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaEventSink.java

  • Create or modify: modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java

  • Create or modify: modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/KafkaEventSinkTest.java

  • Step 1: Write topic routing tests

Create modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/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:

@Test
void rawArchiveRoutesToVersionedGb32960RawTopic() {
    KafkaSinkProperties.Topics topics = new KafkaSinkProperties.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:

@Test
void realtimeRoutesToVersionedGb32960EventTopic() {
    KafkaSinkProperties.Topics topics = new KafkaSinkProperties.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:

mvn -pl :sink-kafka -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 KafkaSinkProperties.Topics defaults

In KafkaSinkProperties.Topics, change defaults:

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:

@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-kafka tests

Run:

mvn -pl :sink-kafka test

Expected: PASS.

  • Step 6: Commit
git add modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java \
        modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/TopicRouter.java \
        modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaEventSink.java \
        modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/TopicRouterTest.java \
        modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/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:

@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:

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:

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:

private final List<EventSink> sinks;

Set it in the constructor:

this.sinks = List.copyOf(sinks);
  • Step 4: Add await API to Dispatcher

In Dispatcher, add:

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):

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:

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:

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:

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:

mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test

Expected: PASS.

  • Step 7: Run core dispatcher tests

Run:

mvn -pl :ingest-core test

Expected: PASS.

  • Step 8: Commit
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:

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:

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:

@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:

mvn -pl :event-history-service,:vehicle-state-service,:vehicle-stat-service test

Expected: PASS.

  • Step 6: Commit
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:

private static Properties loadApplicationProperties() {
    YamlPropertiesFactoryBean yaml = new YamlPropertiesFactoryBean();
    yaml.setResources(new ClassPathResource("application.yml"));
    Properties props = yaml.getObject();
    assertThat(props).isNotNull();
    return props;
}

Imports:

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:

@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.kafka.enabled")).isEqualTo("${KAFKA_ENABLED:true}");
    assertThat(props.getProperty("lingniu.ingest.sink.kafka.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:

@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.kafka.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.group-id"))
            .isEqualTo("${KAFKA_GROUP_HISTORY:vehicle-history}");
}
  • Step 4: Assert analytics defaults

Create VehicleAnalyticsAppDefaultsTest:

@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.kafka.consumer.bindings.vehicleStatEnvelopeConsumerProcessor.group-id"))
            .isEqualTo("${KAFKA_GROUP_STAT:vehicle-stat}");
}
  • Step 5: Run app default tests

Run:

mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test

Expected: PASS.

  • Step 6: Commit
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:

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:

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:

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

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

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:

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:

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:

# 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
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:

mvn -pl :sink-kafka,:ingest-core,:protocol-gb32960,:event-history-service,:vehicle-state-service,:vehicle-stat-service test

Expected: BUILD SUCCESS.

  • Step 2: Run split app tests

Run:

mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test

Expected: BUILD SUCCESS.

  • Step 3: Run package verification

Run:

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:

## 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
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