fix: prepare native cluster deployment

server/kafka_runtime.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"errors"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
@@ -88,6 +89,8 @@ type kafkaRuntime struct {
 
 var kafkaBus *kafkaRuntime
 
+var kafkaConsumerGroupUnsafeChars = regexp.MustCompile(`[^A-Za-z0-9._-]+`)
+
 func initKafkaRuntime(raw json.RawMessage) error {
 	if len(raw) == 0 {
 		kafkaBus = nil
@@ -272,6 +275,19 @@ func publishKafkaEnvelope(ctx context.Context, topicName string, key string, env
 type kafkaMessageHandler func(context.Context, kafkaEnvelope, kafka.Message) error
 
 func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
+	startKafkaConsumerWithGroup(topicName, "", handler)
+}
+
+func startKafkaBroadcastConsumer(topicName string, handler kafkaMessageHandler) string {
+	if kafkaBus == nil || handler == nil {
+		return ""
+	}
+	groupID := kafkaBroadcastConsumerGroup(topicName)
+	startKafkaConsumerWithGroup(topicName, groupID, handler)
+	return groupID
+}
+
+func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessageHandler) {
 	if kafkaBus == nil || handler == nil {
 		return
 	}
@@ -279,8 +295,11 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
 	if topic == "" {
 		topic = topicName
 	}
-	groupID := kafkaBus.config.ConsumerGroup
-	if topicName != "" {
+	groupOverride := groupID != ""
+	if groupID == "" {
+		groupID = kafkaBus.config.ConsumerGroup
+	}
+	if topicName != "" && !groupOverride {
 		groupID += "." + topicName
 	}
 	reader := kafka.NewReader(kafka.ReaderConfig{
@@ -324,6 +343,22 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
 	logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
 }
 
+func kafkaBroadcastConsumerGroup(topicName string) string {
+	if kafkaBus == nil {
+		return ""
+	}
+	base := strings.TrimSpace(kafkaBus.config.ConsumerGroup)
+	if base == "" {
+		base = "im-server"
+	}
+	nodeID := strings.TrimSpace(currentNodeID())
+	if nodeID == "" {
+		nodeID = "local"
+	}
+	nodeID = kafkaConsumerGroupUnsafeChars.ReplaceAllString(nodeID, "-")
+	return base + "." + topicName + "." + nodeID
+}
+
 func (rt *kafkaRuntime) dialer() *kafka.Dialer {
 	dialer := &kafka.Dialer{
 		ClientID: rt.config.ClientID,