add gen+sink configs initialization

pipelane · Jul 31, 2024 · 0f378dd · 0f378dd
1 parent 46caafe
commit 0f378dd
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/source/generator/kafka/consumer.go b/source/generator/kafka/consumer.go
@@ -2,12 +2,14 @@ package kafka
 
 import (
 	"errors"
-	kcfg "github.com/pipelane/pipelaner/source/shared/kafka"
 	"time"
 
+	kcfg "github.com/pipelane/pipelaner/source/shared/kafka"
+
 	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
-	"github.com/pipelane/pipelaner"
 	"github.com/rs/zerolog"
+
+	"github.com/pipelane/pipelaner"
 )
 
 type Kafka struct {
@@ -45,6 +47,7 @@ func NewConsumer(cfg *kcfg.KafkaConfig) (*kafka.Consumer, error) {
 }
 
 func (c *Kafka) Init(ctx *pipelaner.Context) error {
+	c.cfg = new(kcfg.KafkaConfig)
 	err := ctx.LaneItem().Config().ParseExtended(c.cfg)
 	if err != nil {
 		return err

diff --git a/source/sink/clickhouse/clickhouse.go b/source/sink/clickhouse/clickhouse.go
@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+
 	"github.com/huandu/go-sqlbuilder"
-	"github.com/pipelane/pipelaner"
 	"github.com/rs/zerolog"
+
+	"github.com/pipelane/pipelaner"
 )
 
 type Clickhouse struct {
@@ -17,6 +19,7 @@ type Clickhouse struct {
 
 func (c *Clickhouse) Init(ctx *pipelaner.Context) error {
 	c.logger = pipelaner.NewLogger()
+	c.clickConfig = new(ClickhouseConfig)
 	err := ctx.LaneItem().Config().ParseExtended(c.clickConfig)
 	if err != nil {
 		return err

diff --git a/source/sink/kafka/kafka.go b/source/sink/kafka/kafka.go
@@ -6,10 +6,12 @@ package kafka
 
 import (
 	"encoding/json"
+
 	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"github.com/rs/zerolog"
+
 	"github.com/pipelane/pipelaner"
 	kCfg "github.com/pipelane/pipelaner/source/shared/kafka"
-	"github.com/rs/zerolog"
 )
 
 const timeout = 15 * 1000
@@ -22,6 +24,7 @@ type Kafka struct {
 
 func (k *Kafka) Init(ctx *pipelaner.Context) error {
 	k.logger = pipelaner.NewLogger()
+	k.cfg = new(kCfg.KafkaConfig)
 	err := ctx.LaneItem().Config().ParseExtended(k.cfg)
 	if err != nil {
 		return err