Add TLS config for Kafka (#956)

* feat: add TLS utility.

* chore: rename pkg/tls/tls.go to pkg/common/tls/tls.go .

* feat: add util for kafka TLS config.

* feat: setup TLS config for kafka consumer.

* feat: add TLS config to kafka consumer group.

* feat: add TLS config for kafka producer.

* chore: add TLS config for kafka.

* feat: add TLS config for kafka checker.
This commit is contained in:
Wang Zhi
2023-08-25 21:00:53 +08:00
committed by GitHub
parent 6ca1afe641
commit 303162f8a5
7 changed files with 117 additions and 8 deletions
+12 -5
View File
@@ -17,6 +17,7 @@ package kafka
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/tools/log"
"github.com/Shopify/sarama"
@@ -35,11 +36,17 @@ type MConsumerGroupConfig struct {
}
func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []string, groupID string) *MConsumerGroup {
config := sarama.NewConfig()
config.Version = consumerConfig.KafkaVersion
config.Consumer.Offsets.Initial = consumerConfig.OffsetsInitial
config.Consumer.Return.Errors = consumerConfig.IsReturnErr
consumerGroup, err := sarama.NewConsumerGroup(addrs, groupID, config)
consumerGroupConfig := sarama.NewConfig()
consumerGroupConfig.Version = consumerConfig.KafkaVersion
consumerGroupConfig.Consumer.Offsets.Initial = consumerConfig.OffsetsInitial
consumerGroupConfig.Consumer.Return.Errors = consumerConfig.IsReturnErr
if config.Config.Kafka.Username != "" && config.Config.Kafka.Password != "" {
consumerGroupConfig.Net.SASL.Enable = true
consumerGroupConfig.Net.SASL.User = config.Config.Kafka.Username
consumerGroupConfig.Net.SASL.Password = config.Config.Kafka.Password
}
SetupTLSConfig(consumerGroupConfig)
consumerGroup, err := sarama.NewConsumerGroup(addrs, groupID, consumerGroupConfig)
if err != nil {
panic(err.Error())
}