Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feature: add ping for kafka #199

Merged
merged 6 commits into from
Aug 20, 2023
Merged

feature: add ping for kafka #199

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
27a1780
add ping for kafka
ArtARTs36 Aug 20, 2023
31b3174
move ping into FromPipeline/FromConfig
ArtARTs36 Aug 20, 2023
ac533e8
set default ping
ArtARTs36 Aug 20, 2023
8cc9c87
simple
ArtARTs36 Aug 20, 2023
7ffb9f7
add method "pingKafka"
ArtARTs36 Aug 20, 2023
27a1c7d
rename ping method & change log operation name
ArtARTs36 Aug 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions kafkajobs/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ import (
"github.com/twmb/franz-go/pkg/sasl/scram"
)

const defaultPingTimeout = time.Second * 10

func (c *config) InitDefault() ([]kgo.Opt, error) {
opts := make([]kgo.Opt, 0, 1)

Expand Down Expand Up @@ -228,5 +230,11 @@ func (c *config) InitDefault() ([]kgo.Opt, error) {
}
}

if c.Ping == nil {
c.Ping = &Ping{
Timeout: defaultPingTimeout,
}
}

return opts, nil
}
26 changes: 26 additions & 0 deletions kafkajobs/driver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,11 @@ func FromConfig(tracer *sdktrace.TracerProvider, configKey string, log *zap.Logg
return nil, errors.E(op, err)
}

err = pingKafka(jb.kafkaClient, log, conf.Ping.Timeout, pipeline)
if err != nil {
return nil, err
}

jb.pipeline.Store(&pipeline)

go jb.recordsHandler()
Expand Down Expand Up @@ -216,6 +221,11 @@ func FromPipeline(tracer *sdktrace.TracerProvider, pipeline jobs.Pipeline, log *
return nil, errors.E(op, err)
}

err = pingKafka(jb.kafkaClient, log, conf.Ping.Timeout, pipeline)
if err != nil {
return nil, err
}

jb.pipeline.Store(&pipeline)

go jb.recordsHandler()
Expand Down Expand Up @@ -481,3 +491,19 @@ func (d *Driver) requeueHandler() {
}
}
}

func pingKafka(client *kgo.Client, log *zap.Logger, timeout time.Duration, pipe jobs.Pipeline) error {
rustatian marked this conversation as resolved.
Show resolved Hide resolved
const op = errors.Op("new_kafka_consumer")
rustatian marked this conversation as resolved.
Show resolved Hide resolved

pingCtx, pingCancel := context.WithTimeout(context.Background(), timeout)
defer pingCancel()

err := client.Ping(pingCtx)
if err != nil {
return errors.E(op, errors.Errorf("ping kafka was failed: %s", err))
}

log.Debug("ping kafka: ok", zap.String("driver", pipe.Driver()), zap.String("pipeline", pipe.Name()))

return nil
}
5 changes: 5 additions & 0 deletions kafkajobs/opts.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ type config struct {
// global
Brokers []string `mapstructure:"brokers"`
SASL *SASL `mapstructure:"sasl"`
Ping *Ping `mapstructure:"ping"`
rustatian marked this conversation as resolved.
Show resolved Hide resolved

// pipeline
Priority int `mapstructure:"priority"`
Expand All @@ -91,6 +92,10 @@ type SASL struct {
UserAgent string `mapstructure:"user_agent" json:"user_agent"`
}

type Ping struct {
Timeout time.Duration `mapstructure:"timeout" json:"timeout"`
rustatian marked this conversation as resolved.
Show resolved Hide resolved
}

type GroupOptions struct {
GroupID string `mapstructure:"group_id" json:"group_id"`
BlockRebalanceOnPoll bool `mapstructure:"block_rebalance_on_poll" json:"block_rebalance_on_poll"`
Expand Down