chore: scope kafka topics in portainer stack

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}

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"));