458 lines
16 KiB
Markdown
458 lines
16 KiB
Markdown
# JT808 Kafka Streaming Mileage 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:** Consume only JT808 Kafka location events, calculate daily mileage in real time, keep rolling state in Redis or memory, and upsert final/current daily results into MySQL.
|
|
|
|
**Architecture:** Reuse `vehicle-analytics-app` and `vehicle-stat-service` instead of adding a new runtime. The Kafka consumer subscribes to `vehicle.event.jt808.v1`, extracts JT808 location points from `VehicleEnvelope.telemetry_snapshot`, updates one rolling state per `(vehicleKey, statDate)`, stores that state in Redis DB 50 when available, and writes the current daily aggregate to MySQL table `vehicle_daily_mileage_jt808`. No GB32960 comparison and no historical backfill are required for v1.
|
|
|
|
**Tech Stack:** Java 21, Spring Boot auto-configuration, existing `sink-mq` Kafka consumer, protobuf `VehicleEnvelope`, Spring Data Redis `StringRedisTemplate`, Spring JDBC `JdbcTemplate`, MySQL `INSERT ... ON DUPLICATE KEY UPDATE`.
|
|
|
|
---
|
|
|
|
## Confirmed Runtime Inputs
|
|
|
|
- Kafka topic: `vehicle.event.jt808.v1`
|
|
- Statistics source: JT808 only
|
|
- MySQL database: reuse the configured production vehicle database
|
|
- MySQL user: use the configured production vehicle database user
|
|
- Redis endpoint: configure through `REDIS_HOST` and `REDIS_PORT`
|
|
- Redis DB: use `50` by default; `50-60` are available
|
|
- Date zone: `Asia/Shanghai`
|
|
- Segment rules:
|
|
- max segment gap: `PT5M`
|
|
- max implied speed: `200 km/h`
|
|
- short-gap jump filter: `PT10S` and `300 m`
|
|
- Historical backfill: not needed for v1
|
|
- Vehicle key:
|
|
- if VIN is present and not `unknown`, use VIN
|
|
- otherwise use `jt808:{phone}`
|
|
|
|
## Configuration To Add
|
|
|
|
Add these deployment properties to `vehicle-analytics-app`:
|
|
|
|
```yaml
|
|
spring:
|
|
data:
|
|
redis:
|
|
host: ${REDIS_HOST:}
|
|
port: ${REDIS_PORT:6379}
|
|
password: ${REDIS_PASSWORD:}
|
|
database: ${REDIS_DATABASE:50}
|
|
|
|
lingniu:
|
|
ingest:
|
|
sink:
|
|
mq:
|
|
consumer:
|
|
bindings:
|
|
vehicleStatEnvelopeConsumerProcessor:
|
|
topics:
|
|
- ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
|
|
vehicle-stat:
|
|
jt808-mileage:
|
|
enabled: ${JT808_MILEAGE_ENABLED:true}
|
|
mysql:
|
|
jdbc-url: ${JT808_MILEAGE_MYSQL_JDBC_URL:}
|
|
username: ${JT808_MILEAGE_MYSQL_USERNAME:}
|
|
password: ${JT808_MILEAGE_MYSQL_PASSWORD:}
|
|
table-name: ${JT808_MILEAGE_MYSQL_TABLE:vehicle_daily_mileage_jt808}
|
|
initialize-schema: ${JT808_MILEAGE_MYSQL_INITIALIZE_SCHEMA:true}
|
|
state:
|
|
store: ${JT808_MILEAGE_STATE_STORE:redis}
|
|
redis-key-prefix: ${JT808_MILEAGE_REDIS_KEY_PREFIX:vehicle:mileage:jt808:daily:}
|
|
ttl: ${JT808_MILEAGE_REDIS_TTL:P3D}
|
|
max-segment-gap: ${JT808_MILEAGE_MAX_SEGMENT_GAP:PT5M}
|
|
max-implied-speed-kmh: ${JT808_MILEAGE_MAX_IMPLIED_SPEED_KMH:200}
|
|
short-gap: ${JT808_MILEAGE_SHORT_GAP:PT10S}
|
|
short-gap-jump-meters: ${JT808_MILEAGE_SHORT_GAP_JUMP_METERS:300}
|
|
```
|
|
|
|
Do not commit real passwords to `application.yml`; pass them through Nacos or environment variables.
|
|
|
|
## MySQL Table
|
|
|
|
Create automatically when `initialize-schema=true`:
|
|
|
|
```sql
|
|
CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_jt808 (
|
|
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
|
stat_date DATE NOT NULL,
|
|
vehicle_key VARCHAR(64) NOT NULL,
|
|
vin VARCHAR(32) NOT NULL DEFAULT '',
|
|
phone VARCHAR(32) NOT NULL DEFAULT '',
|
|
first_event_time DATETIME(3) NULL,
|
|
last_event_time DATETIME(3) NULL,
|
|
gps_mileage_km DECIMAL(12,3) NOT NULL DEFAULT 0,
|
|
speed_integral_km DECIMAL(12,3) NOT NULL DEFAULT 0,
|
|
odometer_mileage_km DECIMAL(12,3) NULL,
|
|
accepted_points INT NOT NULL DEFAULT 0,
|
|
bad_jump_segments INT NOT NULL DEFAULT 0,
|
|
long_gap_segments INT NOT NULL DEFAULT 0,
|
|
out_of_order_points INT NOT NULL DEFAULT 0,
|
|
odometer_anomalies INT NOT NULL DEFAULT 0,
|
|
data_quality VARCHAR(16) NOT NULL DEFAULT 'PARTIAL',
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
UNIQUE KEY uk_jt808_daily_mileage (stat_date, vehicle_key),
|
|
KEY idx_jt808_daily_mileage_vin_date (vin, stat_date),
|
|
KEY idx_jt808_daily_mileage_phone_date (phone, stat_date)
|
|
);
|
|
```
|
|
|
|
## File Structure
|
|
|
|
- Modify `modules/services/vehicle-stat-service/pom.xml`
|
|
- Add `spring-boot-starter-data-redis`, `spring-jdbc`, and `mysql-connector-j` if not already present through module dependencies.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageProperties.java`
|
|
- Configuration properties for MySQL, Redis state, and GPS filters.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java`
|
|
- Normalized JT808 point.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java`
|
|
- Extracts only JT808 location snapshots from Kafka envelopes.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java`
|
|
- Haversine segment calculator with filtering.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java`
|
|
- Rolling mutable daily state.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java`
|
|
- Persisted result model.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java`
|
|
- State load/save abstraction.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java`
|
|
- Redis JSON state store using `StringRedisTemplate`.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java`
|
|
- Local fallback state store.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java`
|
|
- MySQL schema initialization and upsert.
|
|
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java`
|
|
- One-envelope processing flow.
|
|
- Modify `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java`
|
|
- Fan out to existing odometer processor and optional JT808 mileage stream processor.
|
|
- Modify `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java`
|
|
- Wire the new JT808 mileage beans.
|
|
- Modify `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
|
|
- Change stat binding topic to `vehicle.event.jt808.v1` when JT808 mileage is enabled.
|
|
|
|
## Task 1: JT808 Point Extraction
|
|
|
|
**Files:**
|
|
- Create: `.../jt808/Jt808LocationPoint.java`
|
|
- Create: `.../jt808/Jt808LocationPointExtractor.java`
|
|
- Test: `.../Jt808LocationPointExtractorTest.java`
|
|
|
|
- [ ] **Step 1: Write failing tests**
|
|
|
|
Test cases:
|
|
- Extracts a `source=JT808`, `eventType=LOCATION` envelope with `longitude`, `latitude`, `speed_kmh`, `location_status_raw`, and `total_mileage_km`.
|
|
- Returns empty for `source=GB32960`.
|
|
- Returns empty when longitude or latitude is missing.
|
|
- Uses VIN as `vehicleKey` when VIN is not blank and not `unknown`.
|
|
- Uses `jt808:{phone}` when VIN is `unknown`.
|
|
|
|
- [ ] **Step 2: Implement extractor**
|
|
|
|
The normalized record should contain:
|
|
|
|
```java
|
|
public record Jt808LocationPoint(
|
|
String vehicleKey,
|
|
String vin,
|
|
String phone,
|
|
Instant eventTime,
|
|
double longitude,
|
|
double latitude,
|
|
Double speedKmh,
|
|
Long statusFlag,
|
|
Double totalMileageKm
|
|
) {}
|
|
```
|
|
|
|
- [ ] **Step 3: Verify**
|
|
|
|
Run:
|
|
|
|
```bash
|
|
mvn -pl :vehicle-stat-service -Dtest=Jt808LocationPointExtractorTest test
|
|
```
|
|
|
|
Expected: tests pass.
|
|
|
|
## Task 2: GPS Segment Calculator
|
|
|
|
**Files:**
|
|
- Create: `.../jt808/Jt808GpsMileageCalculator.java`
|
|
- Test: `.../Jt808GpsMileageCalculatorTest.java`
|
|
|
|
- [ ] **Step 1: Write failing tests**
|
|
|
|
Cover:
|
|
- Valid adjacent points accumulate Haversine distance.
|
|
- `dt <= 0` increments out-of-order/non-positive counter.
|
|
- `dt > PT5M` increments long-gap counter and does not accumulate.
|
|
- implied speed `> 200 km/h` increments bad-jump counter and does not accumulate.
|
|
- speed integral uses adjacent average speed.
|
|
|
|
- [ ] **Step 2: Implement calculator**
|
|
|
|
Return a segment result with:
|
|
- `gpsKm`
|
|
- `speedIntegralKm`
|
|
- `used`
|
|
- `badJump`
|
|
- `longGap`
|
|
- `outOfOrder`
|
|
|
|
- [ ] **Step 3: Verify**
|
|
|
|
Run:
|
|
|
|
```bash
|
|
mvn -pl :vehicle-stat-service -Dtest=Jt808GpsMileageCalculatorTest test
|
|
```
|
|
|
|
Expected: tests pass.
|
|
|
|
## Task 3: Rolling State In Redis/Memory
|
|
|
|
**Files:**
|
|
- Create: `.../jt808/Jt808DailyMileageState.java`
|
|
- Create: `.../jt808/Jt808MileageStateStore.java`
|
|
- Create: `.../jt808/RedisJt808MileageStateStore.java`
|
|
- Create: `.../jt808/InMemoryJt808MileageStateStore.java`
|
|
- Test: `.../Jt808DailyMileageStateTest.java`
|
|
- Test: `.../RedisJt808MileageStateStoreTest.java`
|
|
|
|
- [ ] **Step 1: Write failing state tests**
|
|
|
|
Cover:
|
|
- First point initializes state without distance.
|
|
- Ordered point updates GPS and speed mileage.
|
|
- Duplicate/older point is ignored.
|
|
- Monotonic `totalMileageKm` updates `odometerMileageKm`.
|
|
- Odometer rollback increments anomaly counter.
|
|
|
|
- [ ] **Step 2: Implement state key**
|
|
|
|
Use Redis key:
|
|
|
|
```text
|
|
vehicle:mileage:jt808:daily:{statDate}:{vehicleKey}
|
|
```
|
|
|
|
Example:
|
|
|
|
```text
|
|
vehicle:mileage:jt808:daily:2026-06-30:LKLG7C4E7NA774755
|
|
vehicle:mileage:jt808:daily:2026-06-30:jt808:40692934289
|
|
```
|
|
|
|
- [ ] **Step 3: Implement Redis store**
|
|
|
|
Use JSON value and TTL `P3D`. Use DB selection through Spring Redis database config, not by issuing `SELECT` manually.
|
|
|
|
- [ ] **Step 4: Verify**
|
|
|
|
Run:
|
|
|
|
```bash
|
|
mvn -pl :vehicle-stat-service -Dtest=Jt808DailyMileageStateTest,RedisJt808MileageStateStoreTest test
|
|
```
|
|
|
|
Expected: tests pass.
|
|
|
|
## Task 4: MySQL Result Repository
|
|
|
|
**Files:**
|
|
- Create: `.../jt808/Jt808DailyMileageResult.java`
|
|
- Create: `.../jt808/JdbcJt808DailyMileageRepository.java`
|
|
- Test: `.../JdbcJt808DailyMileageRepositoryTest.java`
|
|
|
|
- [ ] **Step 1: Write failing repository tests**
|
|
|
|
Use H2 MySQL mode. Cover:
|
|
- Schema initialization creates `vehicle_daily_mileage_jt808`.
|
|
- Upsert inserts first result.
|
|
- Upsert updates same `(stat_date, vehicle_key)`.
|
|
- Query by date and vehicle key returns latest result.
|
|
|
|
- [ ] **Step 2: Implement repository**
|
|
|
|
Use `JdbcTemplate` and:
|
|
|
|
```sql
|
|
INSERT INTO vehicle_daily_mileage_jt808 (...)
|
|
VALUES (...)
|
|
ON DUPLICATE KEY UPDATE
|
|
vin = VALUES(vin),
|
|
phone = VALUES(phone),
|
|
first_event_time = VALUES(first_event_time),
|
|
last_event_time = VALUES(last_event_time),
|
|
gps_mileage_km = VALUES(gps_mileage_km),
|
|
speed_integral_km = VALUES(speed_integral_km),
|
|
odometer_mileage_km = VALUES(odometer_mileage_km),
|
|
accepted_points = VALUES(accepted_points),
|
|
bad_jump_segments = VALUES(bad_jump_segments),
|
|
long_gap_segments = VALUES(long_gap_segments),
|
|
out_of_order_points = VALUES(out_of_order_points),
|
|
odometer_anomalies = VALUES(odometer_anomalies),
|
|
data_quality = VALUES(data_quality)
|
|
```
|
|
|
|
- [ ] **Step 3: Verify**
|
|
|
|
Run:
|
|
|
|
```bash
|
|
mvn -pl :vehicle-stat-service -Dtest=JdbcJt808DailyMileageRepositoryTest test
|
|
```
|
|
|
|
Expected: tests pass.
|
|
|
|
## Task 5: Kafka Processor Fan-Out
|
|
|
|
**Files:**
|
|
- Create: `.../jt808/Jt808MileageStreamProcessor.java`
|
|
- Modify: `.../VehicleStatEnvelopeIngestor.java`
|
|
- Test: `.../Jt808MileageStreamProcessorTest.java`
|
|
- Test: `.../VehicleStatEnvelopeIngestorTest.java`
|
|
|
|
- [ ] **Step 1: Write failing processor tests**
|
|
|
|
Cover:
|
|
- Two JT808 location envelopes update state and upsert MySQL result.
|
|
- GB32960 envelope is ignored.
|
|
- VIN unknown uses phone key.
|
|
- Different natural days create different states.
|
|
|
|
- [ ] **Step 2: Implement processor flow**
|
|
|
|
Flow:
|
|
|
|
```text
|
|
VehicleEnvelope bytes
|
|
-> Jt808LocationPointExtractor
|
|
-> statDate from eventTime at Asia/Shanghai
|
|
-> stateStore.load(key) or new state
|
|
-> state.apply(point, calculator)
|
|
-> stateStore.save(state)
|
|
-> mysqlRepository.upsert(state.toResult())
|
|
```
|
|
|
|
- [ ] **Step 3: Modify ingestor**
|
|
|
|
Keep the old `VehicleStatEventProcessor` path for compatibility, but add optional `Jt808MileageStreamProcessor`. The new processor ignores non-JT808 envelopes internally.
|
|
|
|
- [ ] **Step 4: Verify**
|
|
|
|
Run:
|
|
|
|
```bash
|
|
mvn -pl :vehicle-stat-service -Dtest=Jt808MileageStreamProcessorTest,VehicleStatEnvelopeIngestorTest test
|
|
```
|
|
|
|
Expected: tests pass.
|
|
|
|
## Task 6: Auto-Configuration And App Properties
|
|
|
|
**Files:**
|
|
- Modify: `.../config/VehicleStatAutoConfiguration.java`
|
|
- Modify: `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
|
|
- Test: `.../config/VehicleStatAutoConfigurationTest.java`
|
|
|
|
- [ ] **Step 1: Write failing auto-config tests**
|
|
|
|
Cover:
|
|
- JT808 mileage beans are present when `lingniu.ingest.vehicle-stat.jt808-mileage.enabled=true`.
|
|
- Redis state store is used when `state.store=redis` and `StringRedisTemplate` exists.
|
|
- Memory state store is used when `state.store=memory`.
|
|
- MySQL repository initializes when config is present.
|
|
|
|
- [ ] **Step 2: Wire beans**
|
|
|
|
Create beans for:
|
|
- `Jt808LocationPointExtractor`
|
|
- `Jt808GpsMileageCalculator`
|
|
- `Jt808MileageStateStore`
|
|
- `JdbcJt808DailyMileageRepository`
|
|
- `Jt808MileageStreamProcessor`
|
|
|
|
- [ ] **Step 3: Update analytics app Kafka binding**
|
|
|
|
For `vehicleStatEnvelopeConsumerProcessor`, topic must be:
|
|
|
|
```yaml
|
|
- ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
|
|
```
|
|
|
|
- [ ] **Step 4: Verify**
|
|
|
|
Run:
|
|
|
|
```bash
|
|
mvn -pl :vehicle-stat-service,:vehicle-analytics-app -am test
|
|
```
|
|
|
|
Expected: tests pass.
|
|
|
|
## Task 7: Deployment Smoke Test
|
|
|
|
**Files:**
|
|
- Modify: `deploy/portainer/docker-compose.yml`
|
|
- Modify: `docs/operations/gb32960-service-split-runbook.md`
|
|
|
|
- [ ] **Step 1: Add environment variables**
|
|
|
|
Add to `vehicle-analytics-app`:
|
|
|
|
```yaml
|
|
REDIS_HOST: ${REDIS_HOST:-}
|
|
REDIS_PORT: ${REDIS_PORT:-6379}
|
|
REDIS_DATABASE: ${REDIS_DATABASE:-50}
|
|
REDIS_PASSWORD: ${REDIS_PASSWORD:-}
|
|
KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1}
|
|
JT808_MILEAGE_ENABLED: ${JT808_MILEAGE_ENABLED:-true}
|
|
JT808_MILEAGE_STATE_STORE: ${JT808_MILEAGE_STATE_STORE:-redis}
|
|
JT808_MILEAGE_MYSQL_JDBC_URL: ${JT808_MILEAGE_MYSQL_JDBC_URL:-}
|
|
JT808_MILEAGE_MYSQL_USERNAME: ${JT808_MILEAGE_MYSQL_USERNAME:-}
|
|
JT808_MILEAGE_MYSQL_PASSWORD: ${JT808_MILEAGE_MYSQL_PASSWORD:-}
|
|
```
|
|
|
|
- [ ] **Step 2: Run live smoke**
|
|
|
|
Start analytics app and verify:
|
|
- Redis keys appear under `vehicle:mileage:jt808:daily:*`.
|
|
- MySQL table `vehicle_daily_mileage_jt808` receives/upserts rows.
|
|
- Kafka consumer group `vehicle-stat` advances on `vehicle.event.jt808.v1`.
|
|
|
|
- [ ] **Step 3: Check one vehicle**
|
|
|
|
For a known active 808 vehicle, query:
|
|
|
|
```sql
|
|
SELECT *
|
|
FROM vehicle_daily_mileage_jt808
|
|
WHERE stat_date = CURRENT_DATE
|
|
ORDER BY gps_mileage_km DESC
|
|
LIMIT 20;
|
|
```
|
|
|
|
Expected: rows show increasing `accepted_points`, `last_event_time`, and mileage during the day.
|
|
|
|
## Remaining Needed Support
|
|
|
|
- MySQL password should be supplied through Nacos/env as `JT808_MILEAGE_MYSQL_PASSWORD`; do not commit it.
|
|
- Redis password value should be supplied through Nacos/env as `REDIS_PASSWORD`; do not commit it.
|
|
|
|
## Risks And Decisions
|
|
|
|
- Redis is rolling state, not the source of truth. MySQL is the durable daily result.
|
|
- Memory state is acceptable only for local/dev or emergency Redis outage; restart loses segment continuity until the next point.
|
|
- No historical backfill means the first deployment day only calculates from the consumer start offset unless Kafka group is reset.
|
|
- JT808 GPS-only mileage is an estimate. Rows with high `bad_jump_segments`, high `long_gap_segments`, or no VIN should be marked `PARTIAL`.
|
|
|
|
## Self-Review
|
|
|
|
- Spec coverage: JT808-only Kafka streaming, Redis/memory state, MySQL result persistence, confirmed topic, confirmed daily algorithm defaults, and no backfill are covered.
|
|
- Placeholder scan: The only open item is deliberately listed under Remaining Needed Support: MySQL connection identity if not already inherited from deployment.
|
|
- Type consistency: The plan uses the `Jt808*` class family throughout and removes cross-protocol verifier scope.
|