From 0f1c1f925512f540ad6a7007dca5135d72c6d664 Mon Sep 17 00:00:00 2001
From: richjohnrambo <richjohnrambo@gmail.com>
Date: Fri, 4 Sep 2026 11:57:38 +0800
Subject: [PATCH] fix: enable local kafka docker integration

---
 docker/tinode/config.template | 14 +++++++-------
 server/kafka_runtime.go       |  8 ++++++--
 2 files changed, 13 insertions(+), 9 deletions(-)

diff --git a/docker/tinode/config.template b/docker/tinode/config.template
index ff237c6..d5d3946 100644
--- a/docker/tinode/config.template
+++ b/docker/tinode/config.template
@@ -118,8 +118,8 @@
 	},
 
 	"kafka": {
-		"enabled": true,
-		"brokers": ["64.118.137.182:9094"],
+		"enabled": $KAFKA_ENABLED,
+		"brokers": $KAFKA_BROKERS_JSON,
 		"client_id": "$KAFKA_CLIENT_ID",
 		"consumer_group": "$KAFKA_CONSUMER_GROUP",
 		"dial_timeout_ms": $KAFKA_DIAL_TIMEOUT_MS,
@@ -127,11 +127,11 @@
 		"read_timeout_ms": $KAFKA_READ_TIMEOUT_MS,
 		"required_acks": $KAFKA_REQUIRED_ACKS,
 		"allow_auto_topic_creation": $KAFKA_ALLOW_AUTO_TOPIC_CREATION,
-		"tls_enabled": false,
-		"sasl_enabled": true,
-		"sasl_mechanism": "SCRAM-SHA-512",
-		"sasl_username": "kafka-external",
-		"sasl_password": "gr58tNnlZOVGW3AmMbL2",
+		"tls_enabled": $KAFKA_TLS_ENABLED,
+		"sasl_enabled": $KAFKA_SASL_ENABLED,
+		"sasl_mechanism": "$KAFKA_SASL_MECHANISM",
+		"sasl_username": "$KAFKA_SASL_USERNAME",
+		"sasl_password": "$KAFKA_SASL_PASSWORD",
 		"max_retry_count": $KAFKA_MAX_RETRY_COUNT,
 		"retry_backoff_ms": $KAFKA_RETRY_BACKOFF_MS,
 		"dead_letter_retention_seconds": $KAFKA_DEAD_LETTER_RETENTION_SECONDS,
diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go
index 4f3ca6e..190821e 100644
--- a/server/kafka_runtime.go
+++ b/server/kafka_runtime.go
@@ -274,9 +274,13 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
 	if topic == "" {
 		topic = topicName
 	}
+	groupID := kafkaBus.config.ConsumerGroup
+	if topicName != "" {
+		groupID += "." + topicName
+	}
 	reader := kafka.NewReader(kafka.ReaderConfig{
 		Brokers:     kafkaBus.config.Brokers,
-		GroupID:     kafkaBus.config.ConsumerGroup,
+		GroupID:     groupID,
 		Topic:       topic,
 		StartOffset: kafka.FirstOffset,
 		MinBytes:    1,
@@ -312,7 +316,7 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
 			}
 		}
 	}()
-	logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, kafkaBus.config.ConsumerGroup)
+	logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
 }
 
 func (rt *kafkaRuntime) dialer() *kafka.Dialer {