From 664a0c801bed19bfb0d5cec168be9e32d9243586 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 15:08:57 +0800 Subject: [PATCH] chore: scope kafka topics in portainer stack --- deploy/portainer/docker-compose.yml | 26 ++++++---- .../PortainerComposeResourceLimitsTest.java | 47 +++++++++++++++++++ 2 files changed, 64 insertions(+), 9 deletions(-) diff --git a/deploy/portainer/docker-compose.yml b/deploy/portainer/docker-compose.yml index 52c6c274..f73fe591 100644 --- a/deploy/portainer/docker-compose.yml +++ b/deploy/portainer/docker-compose.yml @@ -17,15 +17,6 @@ x-common-env: &common-env NACOS_USERNAME: ${NACOS_USERNAME:-} NACOS_PASSWORD: ${NACOS_PASSWORD:-} KAFKA_BROKERS: ${KAFKA_BROKERS:-172.17.111.56:9092} - KAFKA_TOPIC_GB32960_EVENT: ${KAFKA_TOPIC_GB32960_EVENT:-vehicle.event.gb32960.v1} - KAFKA_TOPIC_GB32960_RAW: ${KAFKA_TOPIC_GB32960_RAW:-vehicle.raw.gb32960.v1} - KAFKA_TOPIC_GB32960_DLQ: ${KAFKA_TOPIC_GB32960_DLQ:-vehicle.dlq.gb32960.v1} - KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1} - KAFKA_TOPIC_JT808_RAW: ${KAFKA_TOPIC_JT808_RAW:-vehicle.raw.jt808.v1} - KAFKA_TOPIC_JT808_DLQ: ${KAFKA_TOPIC_JT808_DLQ:-vehicle.dlq.jt808.v1} - KAFKA_TOPIC_YUTONG_MQTT_EVENT: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:-vehicle.event.mqtt-yutong.v1} - KAFKA_TOPIC_YUTONG_MQTT_RAW: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:-vehicle.raw.mqtt-yutong.v1} - KAFKA_TOPIC_YUTONG_MQTT_DLQ: ${KAFKA_TOPIC_YUTONG_MQTT_DLQ:-vehicle.dlq.mqtt-yutong.v1} x-identity-mysql-env: &identity-mysql-env VEHICLE_IDENTITY_MYSQL_JDBC_URL: ${MYSQL_JDBC_URL:-} @@ -59,6 +50,9 @@ services: HTTP_PORT: 20100 GB32960_PORT: 32960 KAFKA_NODE_ID: ${KAFKA_NODE_ID:-gb32960-ingest-portainer} + KAFKA_TOPIC_GB32960_EVENT: ${KAFKA_TOPIC_GB32960_EVENT:-vehicle.event.gb32960.v1} + KAFKA_TOPIC_GB32960_RAW: ${KAFKA_TOPIC_GB32960_RAW:-vehicle.raw.gb32960.v1} + KAFKA_TOPIC_GB32960_DLQ: ${KAFKA_TOPIC_GB32960_DLQ:-vehicle.dlq.gb32960.v1} GB32960_AUTH_ENABLED: ${GB32960_AUTH_ENABLED:-false} GB32960_PLATFORM_USER_HYUNDAI: ${GB32960_PLATFORM_USER_HYUNDAI:-Hyundai} GB32960_PLATFORM_PWD_HYUNDAI: ${GB32960_PLATFORM_PWD_HYUNDAI:-} @@ -92,6 +86,9 @@ services: HTTP_PORT: 20400 JT808_PORT: 808 KAFKA_NODE_ID: ${KAFKA_NODE_ID_JT808:-jt808-ingest-portainer} + KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1} + KAFKA_TOPIC_JT808_RAW: ${KAFKA_TOPIC_JT808_RAW:-vehicle.raw.jt808.v1} + KAFKA_TOPIC_JT808_DLQ: ${KAFKA_TOPIC_JT808_DLQ:-vehicle.dlq.jt808.v1} SINK_ARCHIVE_PATH: ${JT808_ARCHIVE_PATH:-/data/archive} ports: - "${JT808_HTTP_PORT:-20400}:20400" @@ -117,6 +114,9 @@ services: YUTONG_MQTT_ENABLED: ${YUTONG_MQTT_ENABLED:-false} YUTONG_MQTT_AUTO_STARTUP: ${YUTONG_MQTT_AUTO_STARTUP:-true} KAFKA_NODE_ID: ${KAFKA_NODE_ID_MQTT:-yutong-mqtt-portainer} + KAFKA_TOPIC_YUTONG_MQTT_EVENT: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:-vehicle.event.mqtt-yutong.v1} + KAFKA_TOPIC_YUTONG_MQTT_RAW: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:-vehicle.raw.mqtt-yutong.v1} + KAFKA_TOPIC_YUTONG_MQTT_DLQ: ${KAFKA_TOPIC_YUTONG_MQTT_DLQ:-vehicle.dlq.mqtt-yutong.v1} YUTONG_MQTT_ENDPOINT_NAME: ${YUTONG_MQTT_ENDPOINT_NAME:-yutong} YUTONG_MQTT_URI: ${YUTONG_MQTT_URI:-} YUTONG_MQTT_TOPIC: "${YUTONG_MQTT_TOPIC:-#}" @@ -158,6 +158,12 @@ services: KAFKA_CONSUMER_CLIENT_ID_PREFIX: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX_HISTORY:-vehicle-history} KAFKA_CONSUMER_MAX_POLL_RECORDS: ${KAFKA_CONSUMER_MAX_POLL_RECORDS_HISTORY:-2000} KAFKA_CONSUMER_MAX_POLL_INTERVAL_MILLIS: ${KAFKA_CONSUMER_MAX_POLL_INTERVAL_MILLIS_HISTORY:-1800000} + KAFKA_TOPIC_GB32960_EVENT: ${KAFKA_TOPIC_GB32960_EVENT:-vehicle.event.gb32960.v1} + KAFKA_TOPIC_GB32960_RAW: ${KAFKA_TOPIC_GB32960_RAW:-vehicle.raw.gb32960.v1} + KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1} + KAFKA_TOPIC_JT808_RAW: ${KAFKA_TOPIC_JT808_RAW:-vehicle.raw.jt808.v1} + KAFKA_TOPIC_YUTONG_MQTT_EVENT: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:-vehicle.event.mqtt-yutong.v1} + KAFKA_TOPIC_YUTONG_MQTT_RAW: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:-vehicle.raw.mqtt-yutong.v1} KAFKA_GROUP_HISTORY: ${KAFKA_GROUP_HISTORY:-vehicle-history} KAFKA_GROUP_HISTORY_GB32960_EVENT: ${KAFKA_GROUP_HISTORY_GB32960_EVENT:-${KAFKA_GROUP_HISTORY:-vehicle-history}-gb32960-event} KAFKA_GROUP_HISTORY_JT808_EVENT: ${KAFKA_GROUP_HISTORY_JT808_EVENT:-${KAFKA_GROUP_HISTORY:-vehicle-history}-jt808-event} @@ -194,6 +200,8 @@ services: KAFKA_CONSUMER_ENABLED: ${KAFKA_CONSUMER_ENABLED_ANALYTICS:-true} KAFKA_CONSUMER_CLIENT_ID_PREFIX: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX_ANALYTICS:-vehicle-analytics} KAFKA_CONSUMER_MAX_POLL_INTERVAL_MILLIS: ${KAFKA_CONSUMER_MAX_POLL_INTERVAL_MILLIS_ANALYTICS:-900000} + KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1} + KAFKA_TOPIC_JT808_DLQ: ${KAFKA_TOPIC_JT808_DLQ:-vehicle.dlq.jt808.v1} KAFKA_GROUP_STAT: ${KAFKA_GROUP_STAT:-vehicle-stat} VEHICLE_STAT_ENABLED: ${VEHICLE_STAT_ENABLED:-true} VEHICLE_STAT_ZONE_ID: ${VEHICLE_STAT_ZONE_ID:-Asia/Shanghai} diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java index 49239334..a6c559ed 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java @@ -333,6 +333,53 @@ class PortainerComposeResourceLimitsTest { .contains("KAFKA_CONSUMER_MAX_POLL_INTERVAL_MILLIS: ${KAFKA_CONSUMER_MAX_POLL_INTERVAL_MILLIS_ANALYTICS:-900000}"); } + @Test + void kafkaTopicEnvironmentIsScopedToServicesThatUseIt() throws IOException { + String compose = Files.readString(repositoryRoot().resolve("deploy/portainer/docker-compose.yml")); + String commonEnv = commonEnvBlock(compose); + String gb32960Service = serviceBlock(compose, "gb32960-ingest-app"); + String jt808Service = serviceBlock(compose, "jt808-ingest-app"); + String mqttService = serviceBlock(compose, "yutong-mqtt-app"); + String historyService = serviceBlock(compose, "vehicle-history-app"); + String analyticsService = serviceBlock(compose, "vehicle-analytics-app"); + + assertThat(commonEnv).doesNotContain("KAFKA_TOPIC_"); + assertThat(gb32960Service) + .contains("KAFKA_TOPIC_GB32960_EVENT: ${KAFKA_TOPIC_GB32960_EVENT:-vehicle.event.gb32960.v1}") + .contains("KAFKA_TOPIC_GB32960_RAW: ${KAFKA_TOPIC_GB32960_RAW:-vehicle.raw.gb32960.v1}") + .contains("KAFKA_TOPIC_GB32960_DLQ: ${KAFKA_TOPIC_GB32960_DLQ:-vehicle.dlq.gb32960.v1}") + .doesNotContain("KAFKA_TOPIC_JT808") + .doesNotContain("KAFKA_TOPIC_YUTONG_MQTT"); + assertThat(jt808Service) + .contains("KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1}") + .contains("KAFKA_TOPIC_JT808_RAW: ${KAFKA_TOPIC_JT808_RAW:-vehicle.raw.jt808.v1}") + .contains("KAFKA_TOPIC_JT808_DLQ: ${KAFKA_TOPIC_JT808_DLQ:-vehicle.dlq.jt808.v1}") + .doesNotContain("KAFKA_TOPIC_GB32960") + .doesNotContain("KAFKA_TOPIC_YUTONG_MQTT"); + assertThat(mqttService) + .contains("KAFKA_TOPIC_YUTONG_MQTT_EVENT: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:-vehicle.event.mqtt-yutong.v1}") + .contains("KAFKA_TOPIC_YUTONG_MQTT_RAW: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:-vehicle.raw.mqtt-yutong.v1}") + .contains("KAFKA_TOPIC_YUTONG_MQTT_DLQ: ${KAFKA_TOPIC_YUTONG_MQTT_DLQ:-vehicle.dlq.mqtt-yutong.v1}") + .doesNotContain("KAFKA_TOPIC_GB32960") + .doesNotContain("KAFKA_TOPIC_JT808"); + assertThat(historyService) + .contains("KAFKA_TOPIC_GB32960_EVENT: ${KAFKA_TOPIC_GB32960_EVENT:-vehicle.event.gb32960.v1}") + .contains("KAFKA_TOPIC_GB32960_RAW: ${KAFKA_TOPIC_GB32960_RAW:-vehicle.raw.gb32960.v1}") + .contains("KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1}") + .contains("KAFKA_TOPIC_JT808_RAW: ${KAFKA_TOPIC_JT808_RAW:-vehicle.raw.jt808.v1}") + .contains("KAFKA_TOPIC_YUTONG_MQTT_EVENT: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:-vehicle.event.mqtt-yutong.v1}") + .contains("KAFKA_TOPIC_YUTONG_MQTT_RAW: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:-vehicle.raw.mqtt-yutong.v1}") + .doesNotContain("KAFKA_TOPIC_GB32960_DLQ") + .doesNotContain("KAFKA_TOPIC_JT808_DLQ") + .doesNotContain("KAFKA_TOPIC_YUTONG_MQTT_DLQ"); + assertThat(analyticsService) + .contains("KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1}") + .contains("KAFKA_TOPIC_JT808_DLQ: ${KAFKA_TOPIC_JT808_DLQ:-vehicle.dlq.jt808.v1}") + .doesNotContain("KAFKA_TOPIC_GB32960") + .doesNotContain("KAFKA_TOPIC_YUTONG_MQTT") + .doesNotContain("KAFKA_TOPIC_JT808_RAW"); + } + @Test void vehicleHistoryUsesTdengineHistoryEnvironmentNames() throws IOException { String compose = Files.readString(repositoryRoot().resolve("deploy/portainer/docker-compose.yml"));