Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix(#234): use unique consumer groups
Browse files Browse the repository at this point in the history
StephanHCB committed Nov 27, 2023
1 parent 12ecf0a commit 9c7b780
Showing 1 changed file with 3 additions and 3 deletions.
6 changes: 3 additions & 3 deletions internal/repository/kafka/kafka.go
Original file line number Diff line number Diff line change
@@ -3,6 +3,7 @@ package kafka
import (
"context"
"errors"
"fmt"
"github.com/IBM/sarama"
"github.com/Interhyp/metadata-service/internal/acorn/config"
"github.com/Interhyp/metadata-service/internal/acorn/repository"
@@ -131,9 +132,7 @@ func (r *Impl) StartReceiveLoop(ctx context.Context) error {
return errors.New("failed to obtain local non-localhost ip address to use for consumer group, did not get an ipv4 address")
}

workerNodeId := ipComponents[2]

groupId = "metadata-worker" + workerNodeId
groupId = fmt.Sprintf("metadata-worker-%s-%s", ipComponents[2], ipComponents[3])
}
topicConfig.ConsumerGroup = &groupId
r.Logging.Logger().Ctx(ctx).Info().Printf("using kafka group id %s for consumer", groupId)
@@ -142,6 +141,7 @@ func (r *Impl) StartReceiveLoop(ctx context.Context) error {
configPreset.Net.TLS.Enable = true
configPreset.Producer.Compression = sarama.CompressionNone
configPreset.MetricRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, "sarama.consumer.")
configPreset.Consumer.Offsets.Initial = sarama.OffsetNewest

consumer, err := aukafka.CreateConsumer[repository.UpdateEvent](ctx, *topicConfig, callback, configPreset)
if err != nil {

0 comments on commit 9c7b780

Please sign in to comment.