Skip to content

Commit

Permalink
fix: pass topics inside target struct
Browse files Browse the repository at this point in the history
Lakshay Kalbhor committed May 6, 2024
1 parent e6514f4 commit f3cca64
Showing 2 changed files with 7 additions and 5 deletions.
1 change: 1 addition & 0 deletions internal/relay/source_pool.go
Original file line number Diff line number Diff line change
@@ -126,6 +126,7 @@ func (sp *SourcePool) SetInitialOffsets(of map[string]map[int32]kgo.Offset) {
// Set the current candidate with initial weight and a placeholder ID. This initial
// weight ensures we resume consuming from where last left off. A real
// healthy node should replace this via background checks
sp.log.Debug("setting initial target node weight", "weight", w)
sp.curCandidate = Server{
Healthy: false,
Weight: w,
11 changes: 6 additions & 5 deletions internal/relay/target.go
Original file line number Diff line number Diff line change
@@ -41,11 +41,12 @@ type Target struct {
// NewTarget returns a new producer relay that handles target Kafka instances.
func NewTarget(globalCtx context.Context, cfg TargetCfg, pCfg ProducerCfg, topics Topics, m *metrics.Set, log *slog.Logger) (*Target, error) {
p := &Target{
cfg: cfg,
pCfg: pCfg,
ctx: globalCtx,
metrics: m,
log: log,
cfg: cfg,
pCfg: pCfg,
ctx: globalCtx,
metrics: m,
log: log,
targetTopics: topics,

batch: make([]*kgo.Record, 0, pCfg.BatchSize),
batchCh: make(chan *kgo.Record),

0 comments on commit f3cca64

Please sign in to comment.