fix: use sourcePool topics in consumer

init.go

@@ -175,7 +175,7 @@ func initTopicsMap(ko *koanf.Koanf) map[string]relay.Topic {
 }
 
 // initKafkaConfig reads the source(s)/target Kafka configuration.
-func initKafkaConfig(ko *koanf.Koanf, topics map[string]relay.Topic) ([]relay.ConsumerGroupCfg, relay.ProducerCfg) {
+func initKafkaConfig(ko *koanf.Koanf) ([]relay.ConsumerGroupCfg, relay.ProducerCfg) {
 	// Read source Kafka config.
 	src := struct {
 		Sources []relay.ConsumerGroupCfg `koanf:"sources"`
@@ -191,16 +191,6 @@ func initKafkaConfig(ko *koanf.Koanf, topics map[string]relay.Topic) ([]relay.Co
 		log.Fatalf("error unmarshalling `sources` config: %v", err)
 	}
 
-	// Pass all the source topics into cg config
-	var srcs = make([]string, 0, len(topics))
-	for _, t := range topics {
-		srcs = append(srcs, t.SourceTopic)
-	}
-
-	for i := range src.Sources {
-		src.Sources[i].Topics = srcs
-	}
-
 	return src.Sources, prod
 }
 

internal/relay/config.go

@@ -42,8 +42,6 @@ type KafkaCfg struct {
 // ConsumerGroupCfg is the consumer group specific config.
 type ConsumerGroupCfg struct {
 	KafkaCfg `koanf:",squash"`
-
-	Topics []string
 }
 
 // ProducerCfg is the Kafka producer config.

internal/relay/source_pool.go

@@ -376,7 +376,7 @@ func (sp *SourcePool) initConsumerGroup(ctx context.Context, cfg ConsumerGroupCf
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(cfg.BootstrapBrokers...),
 		kgo.FetchMaxWait(sp.cfg.ReqTimeout),
-		kgo.ConsumeTopics(cfg.Topics...),
+		kgo.ConsumeTopics(sp.cfg.Topics...),
 		kgo.ConsumerGroup(sp.cfg.GroupID),
 		kgo.InstanceID(sp.cfg.InstanceID),
 		kgo.SessionTimeout(cfg.SessionTimeout),
@@ -412,7 +412,7 @@ func (sp *SourcePool) initConsumerGroup(ctx context.Context, cfg ConsumerGroupCf
 		return nil, err
 	}
 
-	if err := testConnection(cl, cfg.SessionTimeout, cfg.Topics, nil); err != nil {
+	if err := testConnection(cl, cfg.SessionTimeout, sp.cfg.Topics, nil); err != nil {
 		return nil, err
 	}
 
@@ -546,7 +546,7 @@ waitForTopicLag:
 			}
 
 			// Get end offsets of the topics
-			topicOffsets, err := admCl.ListEndOffsets(ctx, s.Config.Topics...)
+			topicOffsets, err := admCl.ListEndOffsets(ctx, sp.cfg.Topics...)
 			if err != nil {
 				sp.log.Error("error fetching offsets", "err", err)
 				return err

main.go

@@ -50,7 +50,7 @@ func main() {
 	topics := initTopicsMap(ko)
 
 	// Initialize the source and target Kafka config.
-	consumerCfgs, prodConfig := initKafkaConfig(ko, topics)
+	consumerCfgs, prodConfig := initKafkaConfig(ko)
 
 	// Initialize the target Kafka (producer) relay.
 	target, err := relay.NewTarget(globalCtx, initTargetConfig(ko), prodConfig, topics, metr, lo)