IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-04-28 14:29:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

merge: pre-release-v3.8.4 (#3623)

Browse Source 
* merge: pre-release-v3.8.4

* merge: v3.8.4
This commit is contained in:
icey-yu
2025-11-25 16:23:22 +08:00
committed by GitHub
parent 07badb162f
commit b8c4b459fa
17 changed files with 1211 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/storage/kafka/sarama.go
+85
View File
@@ -0,0 +1,85 @@
package kafka
import (
"bytes"
"strings"
"github.com/IBM/sarama"
"github.com/openimsdk/tools/errs"
)
func BuildConsumerGroupConfig(conf *Config, initial int64, autoCommitEnable bool) (*sarama.Config, error) {
kfk := sarama.NewConfig()
kfk.Version = sarama.V2_0_0_0
kfk.Consumer.Offsets.Initial = initial
kfk.Consumer.Offsets.AutoCommit.Enable = autoCommitEnable
kfk.Consumer.Return.Errors = false
if conf.Username != "" || conf.Password != "" {
kfk.Net.SASL.Enable = true
kfk.Net.SASL.User = conf.Username
kfk.Net.SASL.Password = conf.Password
}
if conf.TLS.EnableTLS {
tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify)
if err != nil {
return nil, err
}
kfk.Net.TLS.Config = tls
kfk.Net.TLS.Enable = true
}
return kfk, nil
}
func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error) {
cg, err := sarama.NewConsumerGroup(addr, groupID, conf)
if err != nil {
return nil, errs.WrapMsg(err, "NewConsumerGroup failed", "addr", addr, "groupID", groupID, "conf", *conf)
}
return cg, nil
}
func BuildProducerConfig(conf Config) (*sarama.Config, error) {
kfk := sarama.NewConfig()
kfk.Producer.Return.Successes = true
kfk.Producer.Return.Errors = true
kfk.Producer.Partitioner = sarama.NewHashPartitioner
if conf.Username != "" || conf.Password != "" {
kfk.Net.SASL.Enable = true
kfk.Net.SASL.User = conf.Username
kfk.Net.SASL.Password = conf.Password
}
switch strings.ToLower(conf.ProducerAck) {
case "no_response":
kfk.Producer.RequiredAcks = sarama.NoResponse
case "wait_for_local":
kfk.Producer.RequiredAcks = sarama.WaitForLocal
case "wait_for_all":
kfk.Producer.RequiredAcks = sarama.WaitForAll
default:
kfk.Producer.RequiredAcks = sarama.WaitForAll
}
if conf.CompressType == "" {
kfk.Producer.Compression = sarama.CompressionNone
} else {
if err := kfk.Producer.Compression.UnmarshalText(bytes.ToLower([]byte(conf.CompressType))); err != nil {
return nil, errs.WrapMsg(err, "UnmarshalText failed", "compressType", conf.CompressType)
}
}
if conf.TLS.EnableTLS {
tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify)
if err != nil {
return nil, err
}
kfk.Net.TLS.Config = tls
kfk.Net.TLS.Enable = true
}
return kfk, nil
}
func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error) {
producer, err := sarama.NewSyncProducer(addr, conf)
if err != nil {
return nil, errs.WrapMsg(err, "NewSyncProducer failed", "addr", addr, "conf", *conf)
}
return producer, nil
}