Skip to content

Commit

Permalink
bug: fix kafka transport default partitioning (#246)
Browse files Browse the repository at this point in the history
  • Loading branch information
lspgn authored Dec 6, 2023
1 parent 3c00d97 commit 5ba0210
Show file tree
Hide file tree
Showing 2 changed files with 4 additions and 3 deletions.
6 changes: 3 additions & 3 deletions producer/proto/messages.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,6 @@ func (m *ProtoProducerMessage) MarshalText() ([]byte, error) {
}

func (m *ProtoProducerMessage) baseKey(h hash.Hash) {
if m.formatter == nil || len(m.formatter.key) == 0 {
return
}
vfm := reflect.ValueOf(m)
vfm = reflect.Indirect(vfm)

Expand Down Expand Up @@ -72,6 +69,9 @@ func (m *ProtoProducerMessage) baseKey(h hash.Hash) {
}

func (m *ProtoProducerMessage) Key() []byte {
if m.formatter == nil || len(m.formatter.key) == 0 {
return nil
}
h := fnv.New32()
m.baseKey(h)
return h.Sum(nil)
Expand Down
1 change: 1 addition & 0 deletions transport/kafka/kafka.go
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@ func (d *KafkaDriver) Init() error {
kafkaConfig.Producer.MaxMessageBytes = d.kafkaMaxMsgBytes
kafkaConfig.Producer.Flush.Bytes = d.kafkaFlushBytes
kafkaConfig.Producer.Flush.Frequency = d.kafkaFlushFrequency
kafkaConfig.Producer.Partitioner = sarama.NewRoundRobinPartitioner

if d.kafkaCompressionCodec != "" {
/*
Expand Down

0 comments on commit 5ba0210

Please sign in to comment.