feat: refine IM admin management workflows

server/internal/library/queue/kafkamq.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"crypto/sha512"
+	"errors"
 	"hash"
 	"strings"
 	"time"
@@ -270,6 +271,46 @@ func configureKafkaSASL(conf *sarama.Config, connOpt KafkaConfig) {
 	}
 }
 
+// KafkaHealthCheck verifies broker connectivity by creating a short-lived Sarama client.
+func KafkaHealthCheck(ctx context.Context) error {
+	if len(config.Kafka.Address) == 0 {
+		return errors.New("queue kafka address is empty")
+	}
+	kfkVersion, err := parseKafkaVersion(config.Kafka.Version)
+	if err != nil {
+		return err
+	}
+	if !validateVersion(kfkVersion) {
+		kfkVersion = sarama.V2_4_0_0
+	}
+	conf := sarama.NewConfig()
+	conf.Version = kfkVersion
+	conf.ClientID = "hotgo-healthz"
+	conf.Net.DialTimeout = 2 * time.Second
+	conf.Net.ReadTimeout = 2 * time.Second
+	conf.Net.WriteTimeout = 2 * time.Second
+	configureKafkaSASL(conf, KafkaConfig{
+		UserName:  config.Kafka.Username,
+		Password:  config.Kafka.Password,
+		Mechanism: config.Kafka.Mechanism,
+	})
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	client, err := sarama.NewClient(config.Kafka.Address, conf)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = client.Close() }()
+	if len(client.Brokers()) == 0 {
+		return errors.New("no kafka brokers available")
+	}
+	return nil
+}
+
 func scramClientGenerator(hashFunc func() hash.Hash) func() sarama.SCRAMClient {
 	return func() sarama.SCRAMClient {
 		return &xdgSCRAMClient{HashGeneratorFcn: scram.HashGeneratorFcn(hashFunc)}