diff --git a/e2e/kafka_auth/kafka_auth.go b/e2e/kafka_auth/kafka_auth.go
index 23dffc227..a7217397a 100644
--- a/e2e/kafka_auth/kafka_auth.go
+++ b/e2e/kafka_auth/kafka_auth.go
@@ -111,22 +111,23 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
config.ClientCert = "./kafka_auth/certs/client_cert.pem"
}
- kafka_out.NewProducer(config,
+ kafka_out.NewClient(config,
zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)).Sugar(),
)
},
func() {
config := &kafka_in.Config{
- Brokers: c.Brokers,
- Topics: []string{inTopic},
- ConsumerGroup: "test-auth",
- ClientID: "test-auth-in",
- ChannelBufferSize: 256,
- Offset_: kafka_in.OffsetTypeNewest,
- ConsumerMaxProcessingTime_: 200 * time.Millisecond,
- ConsumerMaxWaitTime_: 250 * time.Millisecond,
- SslEnabled: true,
- SslSkipVerify: true,
+ Brokers: c.Brokers,
+ Topics: []string{inTopic},
+ ConsumerGroup: "test-auth",
+ ClientID: "test-auth-in",
+ ChannelBufferSize: 256,
+ Offset_: kafka_in.OffsetTypeNewest,
+ ConsumerMaxWaitTime_: 250 * time.Millisecond,
+ SslEnabled: true,
+ SslSkipVerify: true,
+ SessionTimeout_: 10 * time.Second,
+ AutoCommitInterval_: 1 * time.Second,
}
if tt.sasl.Enabled {
config.SaslEnabled = true
@@ -139,6 +140,10 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
config.ClientKey = "./kafka_auth/certs/client_key.pem"
config.ClientCert = "./kafka_auth/certs/client_cert.pem"
}
+
+ kafka_in.NewClient(config,
+ zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)).Sugar(),
+ )
},
}
diff --git a/e2e/kafka_file/kafka_file.go b/e2e/kafka_file/kafka_file.go
index 495d256bc..41a2b96e7 100644
--- a/e2e/kafka_file/kafka_file.go
+++ b/e2e/kafka_file/kafka_file.go
@@ -1,16 +1,21 @@
package kafka_file
import (
+ "context"
"log"
"path"
"testing"
"time"
- "github.com/Shopify/sarama"
"github.com/ozontech/file.d/cfg"
+ kafka_out "github.com/ozontech/file.d/plugin/output/kafka"
"github.com/ozontech/file.d/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "github.com/twmb/franz-go/pkg/kadm"
+ "github.com/twmb/franz-go/pkg/kgo"
+ "go.uber.org/zap"
+ "go.uber.org/zap/zapcore"
)
// In this test, a message sender is created that generates one message for each partition. These messages are sent Count times.
@@ -41,27 +46,32 @@ func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string)
// Send creates a Partition of messages (one for each partition) and sends them Count times to kafka
func (c *Config) Send(t *testing.T) {
- config := sarama.NewConfig()
- config.Producer.Flush.Frequency = time.Millisecond
- config.Producer.Return.Errors = true
- config.Producer.Return.Successes = true
+ config := &kafka_out.Config{
+ Brokers: c.Brokers,
+ MaxMessageBytes_: 512,
+ BatchSize_: c.Count,
+ }
- producer, err := sarama.NewSyncProducer(c.Brokers, config)
+ client := kafka_out.NewClient(config,
+ zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)).Sugar(),
+ )
+ adminClient := kadm.NewClient(client)
+ _, err := adminClient.CreateTopic(context.TODO(), 1, 1, nil, c.Topics[0])
if err != nil {
- log.Fatalf("failed to create async producer: %s", err.Error())
+ t.Logf("cannot create topic: %s %s", c.Topics[0], err.Error())
}
- msgs := make([]*sarama.ProducerMessage, c.Partition)
- message := sarama.StringEncoder(`{"key":"value"}`)
+ msgs := make([]*kgo.Record, c.Partition)
for i := range msgs {
- msgs[i] = &sarama.ProducerMessage{}
+ msgs[i] = kgo.SliceRecord([]byte(`{"key":"value"}`))
msgs[i].Topic = c.Topics[0]
- msgs[i].Value = message
msgs[i].Partition = int32(i)
}
for i := 0; i < c.Count; i++ {
- if err = producer.SendMessages(msgs); err != nil {
+ result := client.ProduceSync(context.TODO(), msgs...)
+ err := result.FirstErr()
+ if err != nil {
log.Fatalf("failed to send messages: %s", err.Error())
}
}
diff --git a/e2e/split_join/handler.go b/e2e/split_join/handler.go
deleted file mode 100644
index bd102d28e..000000000
--- a/e2e/split_join/handler.go
+++ /dev/null
@@ -1,19 +0,0 @@
-package split_join
-
-import (
- "github.com/Shopify/sarama"
-)
-
-type handlerFunc func(message *sarama.ConsumerMessage)
-
-func (h handlerFunc) Setup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (h handlerFunc) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (h handlerFunc) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
- for msg := range claim.Messages() {
- h(msg)
- session.MarkMessage(msg, "")
- }
- return nil
-}
diff --git a/e2e/split_join/split_join.go b/e2e/split_join/split_join.go
index 74d3be0ab..b06620b46 100644
--- a/e2e/split_join/split_join.go
+++ b/e2e/split_join/split_join.go
@@ -10,9 +10,13 @@ import (
"testing"
"time"
- "github.com/Shopify/sarama"
"github.com/ozontech/file.d/cfg"
+ kafka_in "github.com/ozontech/file.d/plugin/input/kafka"
"github.com/stretchr/testify/require"
+ "github.com/twmb/franz-go/pkg/kadm"
+ "github.com/twmb/franz-go/pkg/kgo"
+ "go.uber.org/zap"
+ "go.uber.org/zap/zapcore"
)
const (
@@ -27,7 +31,7 @@ const (
type Config struct {
inputDir string
- consumer sarama.ConsumerGroup
+ client *kgo.Client
topic string
}
@@ -48,18 +52,23 @@ func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string)
output.Set("brokers", []string{brokerHost})
output.Set("default_topic", c.topic)
- addrs := []string{brokerHost}
- config := sarama.NewConfig()
- config.Consumer.Offsets.Initial = sarama.OffsetOldest
+ config := &kafka_in.Config{
+ Brokers: []string{brokerHost},
+ Topics: []string{c.topic},
+ ConsumerGroup: group,
+ Offset_: kafka_in.OffsetTypeOldest,
+ SessionTimeout_: 10 * time.Second,
+ AutoCommitInterval_: 1 * time.Second,
+ ConsumerMaxWaitTime_: 1 * time.Second,
+ HeartbeatInterval_: 10 * time.Second,
+ }
- admin, err := sarama.NewClusterAdmin(addrs, config)
- r.NoError(err)
- r.NoError(admin.CreateTopic(c.topic, &sarama.TopicDetail{
- NumPartitions: 1,
- ReplicationFactor: 1,
- }, false))
+ c.client = kafka_in.NewClient(config,
+ zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)).Sugar(),
+ )
- c.consumer, err = sarama.NewConsumerGroup(addrs, group, config)
+ adminClient := kadm.NewClient(c.client)
+ _, err := adminClient.CreateTopic(context.TODO(), 1, 1, nil, c.topic)
r.NoError(err)
}
@@ -89,14 +98,18 @@ func (c *Config) Validate(t *testing.T) {
done := make(chan struct{})
go func() {
- r.NoError(c.consumer.Consume(ctx, []string{c.topic}, handlerFunc(func(msg *sarama.ConsumerMessage) {
- strBuilder.Write(msg.Value)
- strBuilder.WriteString("\n")
- gotEvents++
- if gotEvents == expectedEventsCount {
- close(done)
- }
- })))
+ for {
+ fetches := c.client.PollFetches(ctx)
+ fetches.EachError(func(topic string, p int32, err error) {})
+ fetches.EachRecord(func(r *kgo.Record) {
+ strBuilder.Write(r.Value)
+ strBuilder.WriteString("\n")
+ gotEvents++
+ if gotEvents == expectedEventsCount {
+ close(done)
+ }
+ })
+ }
}()
select {
diff --git a/go.mod b/go.mod
index c4b532442..0712d6598 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,6 @@ require (
github.com/ClickHouse/ch-go v0.58.2
github.com/KimMachineGun/automemlimit v0.2.6
github.com/Masterminds/squirrel v1.5.4
- github.com/Shopify/sarama v1.38.1
github.com/alecthomas/kingpin v2.2.6+incompatible
github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137
github.com/alicebob/miniredis/v2 v2.30.5
@@ -32,6 +31,10 @@ require (
github.com/rjeczalik/notify v0.9.3
github.com/satori/go.uuid v1.2.0
github.com/stretchr/testify v1.9.0
+ github.com/twmb/franz-go v1.17.0
+ github.com/twmb/franz-go/pkg/kadm v1.12.0
+ github.com/twmb/franz-go/plugin/kzap v1.1.2
+ github.com/twmb/tlscfg v1.2.1
github.com/valyala/fasthttp v1.48.0
github.com/vitkovskii/insane-json v0.1.7
github.com/xdg-go/scram v1.1.2
@@ -60,9 +63,6 @@ require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/dmarkham/enumer v1.5.8 // indirect
github.com/docker/go-units v0.4.0 // indirect
- github.com/eapache/go-resiliency v1.3.0 // indirect
- github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6 // indirect
- github.com/eapache/queue v1.1.0 // indirect
github.com/emicklei/go-restful/v3 v3.9.0 // indirect
github.com/go-faster/city v1.0.1 // indirect
github.com/go-faster/errors v0.6.1 // indirect
@@ -76,7 +76,6 @@ require (
github.com/godbus/dbus/v5 v5.0.4 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/protobuf v1.5.3 // indirect
- github.com/golang/snappy v0.0.4 // indirect
github.com/google/gnostic v0.5.7-v3refs // indirect
github.com/google/go-cmp v0.6.0 // indirect
github.com/google/gofuzz v1.1.0 // indirect
@@ -88,7 +87,6 @@ require (
github.com/hashicorp/go-secure-stdlib/parseutil v0.1.6 // indirect
github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 // indirect
github.com/hashicorp/go-sockaddr v1.0.2 // indirect
- github.com/hashicorp/go-uuid v1.0.3 // indirect
github.com/hashicorp/go-version v1.6.0 // indirect
github.com/hashicorp/hcl v1.0.0 // indirect
github.com/imdario/mergo v0.3.7 // indirect
@@ -99,11 +97,6 @@ require (
github.com/jackc/pgtype v1.14.0 // indirect
github.com/jackc/puddle v1.3.0 // indirect
github.com/jackc/puddle/v2 v2.2.1 // indirect
- github.com/jcmturner/aescts/v2 v2.0.0 // indirect
- github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
- github.com/jcmturner/gofork v1.7.6 // indirect
- github.com/jcmturner/gokrb5/v8 v8.4.3 // indirect
- github.com/jcmturner/rpc/v2 v2.0.3 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
@@ -123,16 +116,12 @@ require (
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/prometheus/client_model v0.3.0 // indirect
github.com/prometheus/common v0.42.0 // indirect
- github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
github.com/ryanuber/go-glob v1.0.0 // indirect
github.com/segmentio/asm v1.2.0 // indirect
github.com/sirupsen/logrus v1.8.1 // indirect
github.com/smartystreets/goconvey v1.6.4 // indirect
github.com/spf13/pflag v1.0.5 // indirect
- github.com/twmb/franz-go v1.17.0 // indirect
github.com/twmb/franz-go/pkg/kmsg v1.8.0 // indirect
- github.com/twmb/franz-go/plugin/kzap v1.1.2 // indirect
- github.com/twmb/tlscfg v1.2.1 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/xdg-go/pbkdf2 v1.0.0 // indirect
github.com/xdg-go/stringprep v1.0.4 // indirect
diff --git a/go.sum b/go.sum
index e92449826..bc79b75e2 100644
--- a/go.sum
+++ b/go.sum
@@ -7,10 +7,6 @@ github.com/KimMachineGun/automemlimit v0.2.6/go.mod h1:pJhTW/nWJMj6SnWSU2TEKSlCa
github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
github.com/Masterminds/squirrel v1.5.4 h1:uUcX/aBc8O7Fg9kaISIUsHXdKuqehiXAMQTYX8afzqM=
github.com/Masterminds/squirrel v1.5.4/go.mod h1:NNaOrjSoIDfDA40n7sr2tPNZRfjzjA400rg+riTZj10=
-github.com/Shopify/sarama v1.38.1 h1:lqqPUPQZ7zPqYlWpTh+LQ9bhYNu2xJL6k1SJN4WVe2A=
-github.com/Shopify/sarama v1.38.1/go.mod h1:iwv9a67Ha8VNa+TifujYoWGxWnu2kNVAQdSdZ4X2o5g=
-github.com/Shopify/toxiproxy/v2 v2.5.0 h1:i4LPT+qrSlKNtQf5QliVjdP08GyAH8+BUIc9gT0eahc=
-github.com/Shopify/toxiproxy/v2 v2.5.0/go.mod h1:yhM2epWtAmel9CB8r2+L+PCmhH6yH2pITaPAo7jxJl0=
github.com/alecthomas/kingpin v2.2.6+incompatible h1:5svnBTFgJjZvGKyYBtMB0+m5wvrbUHiqye8wRJMlnYI=
github.com/alecthomas/kingpin v2.2.6+incompatible/go.mod h1:59OFYbFVLKQKq+mqrL6Rw5bR0c3ACQaawgXx0QYndlE=
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
@@ -64,12 +60,6 @@ github.com/dmarkham/enumer v1.5.8/go.mod h1:d10o8R3t/gROm2p3BXqTkMt2+HMuxEmWCXzo
github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
-github.com/eapache/go-resiliency v1.3.0 h1:RRL0nge+cWGlxXbUzJ7yMcq6w2XBEr19dCN6HECGaT0=
-github.com/eapache/go-resiliency v1.3.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho=
-github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6 h1:8yY/I9ndfrgrXUbOGObLHKBR4Fl3nZXwM2c7OYTT8hM=
-github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
-github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
-github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
github.com/emicklei/go-restful/v3 v3.9.0 h1:XwGDlfxEnQZzuopoqxwSEllNcCOM9DhhFyhFIIGKwxE=
github.com/emicklei/go-restful/v3 v3.9.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -78,8 +68,6 @@ github.com/euank/go-kmsg-parser v2.0.0+incompatible h1:cHD53+PLQuuQyLZeriD1V/esu
github.com/euank/go-kmsg-parser v2.0.0+incompatible/go.mod h1:MhmAMZ8V4CYH4ybgdRwPr2TU5ThnS43puaKEMpja1uw=
github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
-github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
-github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
github.com/frankban/quicktest v1.14.0 h1:+cqqvzZV87b4adx/5ayVOaYZ2CrvM4ejQvUdBzPPUss=
github.com/frankban/quicktest v1.14.0/go.mod h1:NeW+ay9A/U67EYXNFA1nPE8e/tnQv/09mUdL/ijj8og=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
@@ -141,8 +129,6 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS
github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
-github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/google/gnostic v0.5.7-v3refs h1:FhTMOKj2VhjpouxvWJAV1TL304uMlb9zcDqkl6cEI54=
github.com/google/gnostic v0.5.7-v3refs/go.mod h1:73MKFl6jIHelAJNaBGFzt3SPtZULs9dYrGFt8OiIsHQ=
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -163,8 +149,6 @@ github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
-github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
@@ -188,9 +172,6 @@ github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 h1:kes8mmyCpxJsI7FTwtzRqEy9
github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4=
github.com/hashicorp/go-sockaddr v1.0.2 h1:ztczhD1jLxIRjVejw8gFomI1BQZOe2WoVOu0SyteCQc=
github.com/hashicorp/go-sockaddr v1.0.2/go.mod h1:rB4wwRAUzs07qva3c5SdrY/NEtAUjGlgmH/UkBUC97A=
-github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
-github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
-github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-version v1.6.0 h1:feTTfFNnjP967rlCxM/I9g701jU+RN74YKx2mOkIeek=
github.com/hashicorp/go-version v1.6.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
@@ -252,18 +233,6 @@ github.com/jackc/puddle v1.3.0 h1:eHK/5clGOatcjX3oWGBO/MpxpbHzSwud5EWTSCI+MX0=
github.com/jackc/puddle v1.3.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
-github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
-github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs=
-github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo=
-github.com/jcmturner/dnsutils/v2 v2.0.0/go.mod h1:b0TnjGOvI/n42bZa+hmXL+kFJZsFT7G4t3HTlQ184QM=
-github.com/jcmturner/gofork v1.7.6 h1:QH0l3hzAU1tfT3rZCnW5zXl+orbkNMMRGJfdJjHVETg=
-github.com/jcmturner/gofork v1.7.6/go.mod h1:1622LH6i/EZqLloHfE7IeZ0uEJwMSUyQ/nDd82IeqRo=
-github.com/jcmturner/goidentity/v6 v6.0.1 h1:VKnZd2oEIMorCTsFBnJWbExfNN7yZr3EhJAxwOkZg6o=
-github.com/jcmturner/goidentity/v6 v6.0.1/go.mod h1:X1YW3bgtvwAXju7V3LCIMpY0Gbxyjn/mY9zx4tFonSg=
-github.com/jcmturner/gokrb5/v8 v8.4.3 h1:iTonLeSJOn7MVUtyMT+arAn5AKAPrkilzhGw8wE/Tq8=
-github.com/jcmturner/gokrb5/v8 v8.4.3/go.mod h1:dqRwJGXznQrzw6cWmyo6kH+E7jksEQG/CyVWsJEsJO0=
-github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY=
-github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
@@ -272,8 +241,6 @@ github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7
github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I=
-github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -344,8 +311,6 @@ github.com/opencontainers/runtime-spec v1.0.2 h1:UfAcuLBJB9Coz72x1hgl8O5RVzTdNia
github.com/opencontainers/runtime-spec v1.0.2/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
github.com/pascaldekloe/name v1.0.1 h1:9lnXOHeqeHHnWLbKfH6X98+4+ETVqFqxN09UXSjcMb0=
github.com/pascaldekloe/name v1.0.1/go.mod h1:Z//MfYJnH4jVpQ9wkclwu2I2MkHmXTlT9wR5UZScttM=
-github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
-github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -365,8 +330,6 @@ github.com/prometheus/common v0.42.0 h1:EKsfXEYo4JpWMHH5cg+KOUWeuJSov1Id8zGR8eeI
github.com/prometheus/common v0.42.0/go.mod h1:xBwqVerjNdUDjgODMpudtOMwlOwf2SaTr1yjz4b7Zbc=
github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
-github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
-github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/rjeczalik/notify v0.9.3 h1:6rJAzHTGKXGj76sbRgDiDcYj/HniypXmSJo1SWakZeY=
github.com/rjeczalik/notify v0.9.3/go.mod h1:gF3zSOrafR9DQEWSE8TjfI9NkooDxbyT4UgRGKZA0lc=
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
@@ -414,6 +377,8 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/twmb/franz-go v1.17.0 h1:hawgCx5ejDHkLe6IwAtFWwxi3OU4OztSTl7ZV5rwkYk=
github.com/twmb/franz-go v1.17.0/go.mod h1:NreRdJ2F7dziDY/m6VyspWd6sNxHKXdMZI42UfQ3GXM=
+github.com/twmb/franz-go/pkg/kadm v1.12.0 h1:I8P/gpXFzhl73QcAYmJu+1fOXvrynyH/MAotr2udEg4=
+github.com/twmb/franz-go/pkg/kadm v1.12.0/go.mod h1:VMvpfjz/szpH9WB+vGM+rteTzVv0djyHFimci9qm2C0=
github.com/twmb/franz-go/pkg/kmsg v1.8.0 h1:lAQB9Z3aMrIP9qF9288XcFf/ccaSxEitNA1CDTEIeTA=
github.com/twmb/franz-go/pkg/kmsg v1.8.0/go.mod h1:HzYEb8G3uu5XevZbtU0dVbkphaKTHk0X68N5ka4q6mU=
github.com/twmb/franz-go/plugin/kzap v1.1.2 h1:0arX5xJ0soUPX1LlDay6ZZoxuWkWk1lggQ5M/IgRXAE=
@@ -479,10 +444,7 @@ golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWP
golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
-golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
-golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -509,18 +471,13 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
-golang.org/x/net v0.0.0-20220725212005-46097bf591d3/go.mod h1:AaygXjzTFtRAg2ttMY5RMuhpJ3cNnI0XpyFJD1iQRSM=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
-golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -559,22 +516,17 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
-golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
-golang.org/x/term v0.13.0 h1:bb+I9cTfFazGW51MZqBVmZy7+JEJMouUHTUSKVQLBek=
-golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
golang.org/x/term v0.20.0 h1:VnkxpohqXaOBYJtBmEppKUG6mXpi+4O6purfc2+sMhw=
golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -585,8 +537,6 @@ golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
-golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 h1:vVKdlvoWBphwdxWKrFZEuM0kGgGLxUOYcY4U/2Vjg44=
diff --git a/logger/logger.go b/logger/logger.go
index f11d437b2..9fc54693a 100644
--- a/logger/logger.go
+++ b/logger/logger.go
@@ -5,7 +5,6 @@ import (
"strings"
"time"
- "github.com/Shopify/sarama"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)
@@ -55,14 +54,6 @@ func init() {
Sugar().
Named("fd")
- saramaConfig := config
- // omit level key for sarama logger
- saramaConfig.LevelKey = zapcore.OmitKey
- saramaCore := zapcore.NewCore(zapcore.NewJSONEncoder(saramaConfig), zapcore.AddSync(os.Stderr), Level)
- saramaLogger := zap.New(saramaCore).Named("sarama")
- sarama.Logger, _ = zap.NewStdLogAt(saramaLogger, zapcore.InfoLevel)
- sarama.DebugLogger, _ = zap.NewStdLogAt(saramaLogger, zapcore.DebugLevel)
-
Instance.Infof("Logger initialized with level: %s", level)
}
diff --git a/plugin/README.md b/plugin/README.md
index e31c8496d..792091b5a 100755
--- a/plugin/README.md
+++ b/plugin/README.md
@@ -137,7 +137,7 @@ pipelines:
[More details...](plugin/input/k8s/README.md)
## kafka
-It reads events from multiple Kafka topics using `sarama` library.
+It reads events from multiple Kafka topics using `franz-go` library.
> It guarantees at "at-least-once delivery" due to the commitment mechanism.
**Example**
@@ -722,7 +722,7 @@ Allowed characters in field names are letters, numbers, underscores, dashes, and
[More details...](plugin/output/gelf/README.md)
## kafka
-It sends the event batches to kafka brokers using `sarama` lib.
+It sends the event batches to kafka brokers using `franz-go` lib.
[More details...](plugin/output/kafka/README.md)
## postgres
diff --git a/plugin/input/README.md b/plugin/input/README.md
index 0df508aaa..8afbeb3a2 100755
--- a/plugin/input/README.md
+++ b/plugin/input/README.md
@@ -136,7 +136,7 @@ pipelines:
[More details...](plugin/input/k8s/README.md)
## kafka
-It reads events from multiple Kafka topics using `sarama` library.
+It reads events from multiple Kafka topics using `franz-go` library.
> It guarantees at "at-least-once delivery" due to the commitment mechanism.
**Example**
diff --git a/plugin/input/kafka/README.md b/plugin/input/kafka/README.md
index 4f1800043..d10102365 100755
--- a/plugin/input/kafka/README.md
+++ b/plugin/input/kafka/README.md
@@ -1,5 +1,5 @@
# Kafka plugin
-It reads events from multiple Kafka topics using `sarama` library.
+It reads events from multiple Kafka topics using `franz-go` library.
> It guarantees at "at-least-once delivery" due to the commitment mechanism.
**Example**
diff --git a/plugin/input/kafka/client.go b/plugin/input/kafka/client.go
index 3a36cc2ae..8ddd34e95 100644
--- a/plugin/input/kafka/client.go
+++ b/plugin/input/kafka/client.go
@@ -1,6 +1,8 @@
package kafka
import (
+ "context"
+ "crypto/tls"
"os"
"github.com/twmb/franz-go/pkg/kgo"
@@ -12,7 +14,7 @@ import (
"go.uber.org/zap"
)
-func newClient(c *Config, l *zap.SugaredLogger) *kgo.Client {
+func NewClient(c *Config, l *zap.SugaredLogger) *kgo.Client {
opts := []kgo.Opt{
kgo.SeedBrokers(c.Brokers...),
kgo.ConsumerGroup(c.ConsumerGroup),
@@ -27,11 +29,6 @@ func newClient(c *Config, l *zap.SugaredLogger) *kgo.Client {
kgo.AutoCommitInterval(c.AutoCommitInterval_),
kgo.SessionTimeout(c.SessionTimeout_),
kgo.HeartbeatInterval(c.HeartbeatInterval_),
-
- // kgo.RequestTimeoutOverhead(), TODO: ConsumerMaxProcessingTime?
- // https://github.com/twmb/franz-go/blob/40589af736a73742a24db882c0088669ca9cf0ca/examples/bench/main.go#L191
- // https://github.com/twmb/franz-go/blob/40589af736a73742a24db882c0088669ca9cf0ca/pkg/kgo/config.go#L132-L186
- // https://github.com/twmb/franz-go/blob/master/docs/producing-and-consuming.md#consuming
}
offset := kgo.NewOffset()
@@ -66,35 +63,38 @@ func newClient(c *Config, l *zap.SugaredLogger) *kgo.Client {
SecretKey: c.SaslPassword,
}.AsManagedStreamingIAMMechanism()))
}
+ opts = append(opts, kgo.DialTLSConfig(new(tls.Config)))
}
- if c.CACert != "" || c.ClientCert != "" || c.ClientKey != "" {
+ if c.SslEnabled {
tlsOpts := []tlscfg.Opt{}
+ if c.CACert != "" || c.ClientCert != "" || c.ClientKey != "" {
+ if c.CACert != "" {
+ if _, err := os.Stat(c.CACert); err != nil {
+ tlsOpts = append(tlsOpts,
+ tlscfg.WithCA(
+ []byte(c.CACert), tlscfg.ForClient,
+ ),
+ )
+ } else {
+ tlsOpts = append(tlsOpts,
+ tlscfg.MaybeWithDiskCA(c.CACert, tlscfg.ForClient),
+ )
+ }
+ }
- if _, err := os.Stat(c.CACert); err != nil {
- tlsOpts = append(tlsOpts,
- tlscfg.WithCA(
- []byte(c.CACert), tlscfg.ForClient,
- ),
- )
- } else {
- tlsOpts = append(tlsOpts,
- tlscfg.MaybeWithDiskCA(c.CACert, tlscfg.ForClient),
- )
- }
-
- if _, err := os.Stat(c.ClientCert); err != nil {
- tlsOpts = append(tlsOpts,
- tlscfg.WithKeyPair(
- []byte(c.ClientCert), []byte(c.ClientKey),
- ),
- )
- } else {
- tlsOpts = append(tlsOpts,
- tlscfg.MaybeWithDiskKeyPair(c.ClientCert, c.ClientKey),
- )
+ if _, err := os.Stat(c.ClientCert); err != nil {
+ tlsOpts = append(tlsOpts,
+ tlscfg.WithKeyPair(
+ []byte(c.ClientCert), []byte(c.ClientKey),
+ ),
+ )
+ } else {
+ tlsOpts = append(tlsOpts,
+ tlscfg.MaybeWithDiskKeyPair(c.ClientCert, c.ClientKey),
+ )
+ }
}
-
tc, err := tlscfg.New(tlsOpts...)
if err != nil {
l.Fatalf("unable to create tls config: %v", err)
@@ -115,10 +115,13 @@ func newClient(c *Config, l *zap.SugaredLogger) *kgo.Client {
}
client, err := kgo.NewClient(opts...)
-
if err != nil {
l.Fatalf("can't create kafka client: %s", err.Error())
}
+ err = client.Ping(context.TODO())
+ if err != nil {
+ l.Fatalf("can't connect to kafka: %s", err.Error())
+ }
return client
}
diff --git a/plugin/input/kafka/kafka.go b/plugin/input/kafka/kafka.go
index d4e7cb3c7..554cfc818 100644
--- a/plugin/input/kafka/kafka.go
+++ b/plugin/input/kafka/kafka.go
@@ -16,7 +16,7 @@ import (
)
/*{ introduction
-It reads events from multiple Kafka topics using `sarama` library.
+It reads events from multiple Kafka topics using `franz-go` library.
> It guarantees at "at-least-once delivery" due to the commitment mechanism.
**Example**
@@ -239,7 +239,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.InputPluginPa
ctx, cancel := context.WithCancel(context.Background())
p.cancel = cancel
- p.client = newClient(p.config, p.logger)
+ p.client = NewClient(p.config, p.logger)
p.controller.UseSpread()
p.controller.DisableStreams()
diff --git a/plugin/output/README.md b/plugin/output/README.md
index 9b05cc4ef..e74089fce 100755
--- a/plugin/output/README.md
+++ b/plugin/output/README.md
@@ -41,7 +41,7 @@ Allowed characters in field names are letters, numbers, underscores, dashes, and
[More details...](plugin/output/gelf/README.md)
## kafka
-It sends the event batches to kafka brokers using `sarama` lib.
+It sends the event batches to kafka brokers using `franz-go` lib.
[More details...](plugin/output/kafka/README.md)
## postgres
diff --git a/plugin/output/kafka/README.md b/plugin/output/kafka/README.md
index 4b6d052bb..7800bd958 100755
--- a/plugin/output/kafka/README.md
+++ b/plugin/output/kafka/README.md
@@ -1,5 +1,5 @@
# Kafka output
-It sends the event batches to kafka brokers using `sarama` lib.
+It sends the event batches to kafka brokers using `franz-go` lib.
### Config params
**`brokers`** *`[]string`* *`required`*
@@ -64,12 +64,24 @@ Should be set equal to or smaller than the broker's `message.max.bytes`.
+**`linger`** *`cfg.Duration`* *`default=0`*
+
+Linger sets how long individual topic partitions will linger waiting
+
+
+
**`compression`** *`string`* *`default=none`* *`options=none|gzip|snappy|lz4|zstd`*
Compression
+**`ack`** *`string`* *`default=all-isr`* *`options=no|leader|all-isr`*
+
+Ack
+
+
+
**`retry`** *`int`* *`default=10`*
Retries of insertion. If File.d cannot insert for this number of attempts,
diff --git a/plugin/output/kafka/client.go b/plugin/output/kafka/client.go
new file mode 100644
index 000000000..d3674636d
--- /dev/null
+++ b/plugin/output/kafka/client.go
@@ -0,0 +1,130 @@
+package kafka
+
+import (
+ "context"
+ "crypto/tls"
+ "os"
+
+ "github.com/twmb/franz-go/pkg/kgo"
+ "github.com/twmb/franz-go/pkg/sasl/aws"
+ "github.com/twmb/franz-go/pkg/sasl/plain"
+ "github.com/twmb/franz-go/pkg/sasl/scram"
+ "github.com/twmb/franz-go/plugin/kzap"
+ "github.com/twmb/tlscfg"
+ "go.uber.org/zap"
+)
+
+type KafkaClient interface {
+ ProduceSync(ctx context.Context, rs ...*kgo.Record) kgo.ProduceResults
+ Close()
+}
+
+func NewClient(c *Config, l *zap.SugaredLogger) *kgo.Client {
+ opts := []kgo.Opt{
+ kgo.SeedBrokers(c.Brokers...),
+ kgo.ClientID(c.ClientID),
+ kgo.DefaultProduceTopic(c.DefaultTopic),
+ kgo.WithLogger(kzap.New(l.Desugar())),
+ kgo.MaxBufferedRecords(c.BatchSize_),
+ kgo.ProducerBatchMaxBytes(int32(c.MaxMessageBytes_)),
+ kgo.ProducerLinger(c.Linger_),
+ }
+
+ if c.SaslEnabled {
+ switch c.SaslMechanism {
+ case "PLAIN":
+ opts = append(opts, kgo.SASL(plain.Auth{
+ User: c.SaslUsername,
+ Pass: c.SaslPassword,
+ }.AsMechanism()))
+ case "SCRAM-SHA-256":
+ opts = append(opts, kgo.SASL(scram.Auth{
+ User: c.SaslUsername,
+ Pass: c.SaslPassword,
+ }.AsSha256Mechanism()))
+ case "SCRAM-SHA-512":
+ opts = append(opts, kgo.SASL(scram.Auth{
+ User: c.SaslUsername,
+ Pass: c.SaslPassword,
+ }.AsSha512Mechanism()))
+ case "AWS_MSK_IAM":
+ opts = append(opts, kgo.SASL(aws.Auth{
+ AccessKey: c.SaslUsername,
+ SecretKey: c.SaslPassword,
+ }.AsManagedStreamingIAMMechanism()))
+ }
+ opts = append(opts, kgo.DialTLSConfig(new(tls.Config)))
+ }
+
+ if c.SslEnabled {
+ tlsOpts := []tlscfg.Opt{}
+ if c.CACert != "" || c.ClientCert != "" || c.ClientKey != "" {
+ if c.CACert != "" {
+ if _, err := os.Stat(c.CACert); err != nil {
+ tlsOpts = append(tlsOpts,
+ tlscfg.WithCA(
+ []byte(c.CACert), tlscfg.ForClient,
+ ),
+ )
+ } else {
+ tlsOpts = append(tlsOpts,
+ tlscfg.MaybeWithDiskCA(c.CACert, tlscfg.ForClient),
+ )
+ }
+ }
+
+ if _, err := os.Stat(c.ClientCert); err != nil {
+ tlsOpts = append(tlsOpts,
+ tlscfg.WithKeyPair(
+ []byte(c.ClientCert), []byte(c.ClientKey),
+ ),
+ )
+ } else {
+ tlsOpts = append(tlsOpts,
+ tlscfg.MaybeWithDiskKeyPair(c.ClientCert, c.ClientKey),
+ )
+ }
+ }
+ tc, err := tlscfg.New(tlsOpts...)
+ if err != nil {
+ l.Fatalf("unable to create tls config: %v", err)
+ }
+ tc.InsecureSkipVerify = c.SslSkipVerify
+ opts = append(opts, kgo.DialTLSConfig(tc))
+ }
+
+ switch c.Compression {
+ case "none":
+ opts = append(opts, kgo.ProducerBatchCompression(kgo.NoCompression()))
+ case "gzip":
+ opts = append(opts, kgo.ProducerBatchCompression(kgo.GzipCompression()))
+ case "snappy":
+ opts = append(opts, kgo.ProducerBatchCompression(kgo.SnappyCompression()))
+ case "lz4":
+ opts = append(opts, kgo.ProducerBatchCompression(kgo.Lz4Compression()))
+ case "zstd":
+ opts = append(opts, kgo.ProducerBatchCompression(kgo.ZstdCompression()))
+ }
+
+ switch c.Ack {
+ case "no":
+ opts = append(opts, kgo.RequiredAcks(kgo.NoAck()), kgo.DisableIdempotentWrite())
+ case "leader":
+ opts = append(opts, kgo.RequiredAcks(kgo.LeaderAck()), kgo.DisableIdempotentWrite())
+ case "all-isr":
+ opts = append(opts, kgo.RequiredAcks(kgo.AllISRAcks()))
+ }
+
+ client, err := kgo.NewClient(opts...)
+
+ if err != nil {
+ l.Fatalf("can't create kafka client: %s", err.Error())
+ }
+
+ err = client.Ping(context.TODO())
+ if err != nil {
+ l.Fatalf("can't connect to kafka: %s", err.Error())
+ }
+
+ return client
+}
diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go
index 8953ee3ae..d7f2140c9 100644
--- a/plugin/output/kafka/kafka.go
+++ b/plugin/output/kafka/kafka.go
@@ -2,23 +2,20 @@ package kafka
import (
"context"
- "strings"
"time"
- "github.com/Shopify/sarama"
"github.com/ozontech/file.d/cfg"
"github.com/ozontech/file.d/fd"
"github.com/ozontech/file.d/metric"
"github.com/ozontech/file.d/pipeline"
- "github.com/ozontech/file.d/xscram"
- "github.com/ozontech/file.d/xtls"
"github.com/prometheus/client_golang/prometheus"
+ "github.com/twmb/franz-go/pkg/kgo"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)
/*{ introduction
-It sends the event batches to kafka brokers using `sarama` lib.
+It sends the event batches to kafka brokers using `franz-go` lib.
}*/
const (
@@ -26,8 +23,8 @@ const (
)
type data struct {
- messages []*sarama.ProducerMessage
- outBuf sarama.ByteEncoder
+ messages []*kgo.Record
+ outBuf []byte
}
type Plugin struct {
@@ -36,8 +33,8 @@ type Plugin struct {
avgEventSize int
controller pipeline.OutputPluginController
- producer sarama.SyncProducer
- batcher *pipeline.RetriableBatcher
+ client KafkaClient
+ batcher *pipeline.RetriableBatcher
// plugin metrics
sendErrorMetric prometheus.Counter
@@ -103,11 +100,22 @@ type Config struct {
MaxMessageBytes cfg.Expression `json:"max_message_bytes" default:"1000000" parse:"expression"` // *
MaxMessageBytes_ int
+ // > @3@4@5@6
+ // >
+ // > Linger sets how long individual topic partitions will linger waiting
+ Linger cfg.Duration `json:"linger" default:"0" parse:"duration"` // *
+ Linger_ time.Duration
+
// > @3@4@5@6
// >
// > Compression
Compression string `json:"compression" default:"none" options:"none|gzip|snappy|lz4|zstd"` // *
+ // > @3@4@5@6
+ // >
+ // > Ack
+ Ack string `json:"ack" default:"all-isr" options:"no|leader|all-isr"` // *
+
// > @3@4@5@6
// >
// > Retries of insertion. If File.d cannot insert for this number of attempts,
@@ -201,7 +209,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
p.logger.Infof("workers count=%d, batch size=%d", p.config.WorkersCount_, p.config.BatchSize_)
- p.producer = NewProducer(p.config, p.logger)
+ p.client = NewClient(p.config, p.logger)
batcherOpts := pipeline.BatcherOptions{
PipelineName: params.PipelineName,
@@ -254,7 +262,7 @@ func (p *Plugin) registerMetrics(ctl *metric.Ctl) {
func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) error {
if *workerData == nil {
*workerData = &data{
- messages: make([]*sarama.ProducerMessage, p.config.BatchSize_),
+ messages: make([]*kgo.Record, p.config.BatchSize_),
outBuf: make([]byte, 0, p.config.BatchSize_*p.avgEventSize),
}
}
@@ -262,7 +270,7 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err
data := (*workerData).(*data)
// handle to much memory consumption
if cap(data.outBuf) > p.config.BatchSize_*p.avgEventSize {
- data.outBuf = make(sarama.ByteEncoder, 0, p.config.BatchSize_*p.avgEventSize)
+ data.outBuf = make([]byte, 0, p.config.BatchSize_*p.avgEventSize)
}
outBuf := data.outBuf[:0]
@@ -280,90 +288,35 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err
}
if data.messages[i] == nil {
- data.messages[i] = &sarama.ProducerMessage{}
+ data.messages[i] = kgo.SliceRecord(outBuf[start:])
}
- data.messages[i].Value = outBuf[start:]
data.messages[i].Topic = topic
i++
})
- err := p.producer.SendMessages(data.messages[:i])
- if err != nil {
- errs := err.(sarama.ProducerErrors)
- for _, e := range errs {
- p.logger.Errorf("can't write batch: %s", e.Err.Error())
- }
- p.sendErrorMetric.Add(float64(len(errs)))
- p.logger.Error(
- "an attempt to insert a batch failed",
- zap.Error(err),
- )
- }
-
- return err
-}
-
-func (p *Plugin) Stop() {
- p.batcher.Stop()
- if err := p.producer.Close(); err != nil {
- p.logger.Error("can't stop kafka producer: %s", err)
- }
-}
-
-func NewProducer(c *Config, l *zap.SugaredLogger) sarama.SyncProducer {
- config := sarama.NewConfig()
- config.ClientID = c.ClientID
-
- // kafka auth sasl
- if c.SaslEnabled {
- config.Net.SASL.Enable = true
-
- config.Net.SASL.User = c.SaslUsername
- config.Net.SASL.Password = c.SaslPassword
-
- config.Net.SASL.Mechanism = sarama.SASLMechanism(c.SaslMechanism)
- switch config.Net.SASL.Mechanism {
- case sarama.SASLTypeSCRAMSHA256:
- config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return xscram.NewClient(xscram.SHA256) }
- case sarama.SASLTypeSCRAMSHA512:
- config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return xscram.NewClient(xscram.SHA512) }
- }
- }
-
- // kafka connect via SSL with PEM
- if c.SslEnabled {
- config.Net.TLS.Enable = true
+ results := p.client.ProduceSync(context.TODO(), data.messages[:i]...)
- tlsCfg := xtls.NewConfigBuilder()
- if c.CACert != "" {
- if err := tlsCfg.AppendCARoot(c.CACert); err != nil {
- l.Fatalf("can't load ca cert: %s", err.Error())
+ if results.FirstErr() != nil {
+ errors := 0
+ for _, result := range results {
+ if result.Err != nil {
+ p.logger.Errorf("can't write batch: %s", result.Err.Error())
+ errors++
}
}
- tlsCfg.SetSkipVerify(c.SslSkipVerify)
-
- if c.ClientCert != "" || c.ClientKey != "" {
- if err := tlsCfg.AppendX509KeyPair(c.ClientCert, c.ClientKey); err != nil {
- l.Fatalf("can't load client certificate and key: %s", err.Error())
- }
+ if errors > 0 {
+ p.sendErrorMetric.Add(float64(errors))
+ p.logger.Error(
+ "an attempt to insert a batch failed",
+ zap.Error(results.FirstErr()),
+ )
}
-
- config.Net.TLS.Config = tlsCfg.Build()
}
- config.Producer.MaxMessageBytes = c.MaxMessageBytes_
- config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
- config.Producer.Flush.Messages = c.BatchSize_
- // kafka plugin itself cares for flush frequency, but we are using batcher so disable it.
- config.Producer.Flush.Frequency = time.Millisecond
- config.Producer.Return.Errors = true
- config.Producer.Return.Successes = true
-
- producer, err := sarama.NewSyncProducer(c.Brokers, config)
- if err != nil {
- l.Fatalf("can't create producer: %s", err.Error())
- }
+ return results.FirstErr()
+}
- l.Infof("producer created with brokers %q", strings.Join(c.Brokers, ","))
- return producer
+func (p *Plugin) Stop() {
+ p.batcher.Stop()
+ p.client.Close()
}
diff --git a/plugin/output/kafka/kafka_test.go b/plugin/output/kafka/kafka_test.go
index 552a494a3..9c4a797c8 100644
--- a/plugin/output/kafka/kafka_test.go
+++ b/plugin/output/kafka/kafka_test.go
@@ -4,12 +4,12 @@
package kafka
import (
+ "context"
"fmt"
"testing"
- "github.com/Shopify/sarama"
"github.com/ozontech/file.d/pipeline"
- "github.com/stretchr/testify/require"
+ "github.com/twmb/franz-go/pkg/kgo"
insaneJSON "github.com/vitkovskii/insane-json"
"go.uber.org/zap/zaptest"
)
@@ -23,61 +23,12 @@ type mockProducer struct {
t *testing.T
}
-func (m *mockProducer) TxnStatus() sarama.ProducerTxnStatusFlag {
- return 0
-}
-
-func (m *mockProducer) IsTransactional() bool {
- return false
-}
-
-func (m *mockProducer) BeginTxn() error {
- return nil
-}
-
-func (m *mockProducer) CommitTxn() error {
+func (m *mockProducer) ProduceSync(ctx context.Context, rs ...*kgo.Record) kgo.ProduceResults {
return nil
}
-func (m *mockProducer) AbortTxn() error {
- return nil
-}
+func (m *mockProducer) Close() {
-func (m *mockProducer) AddOffsetsToTxn(offsets map[string][]*sarama.PartitionOffsetMetadata, groupId string) error {
- return nil
-}
-
-func (m *mockProducer) AddMessageToTxn(msg *sarama.ConsumerMessage, groupId string, metadata *string) error {
- return nil
-}
-
-func (m *mockProducer) ensureTopic(msg *sarama.ProducerMessage) {
- val, err := msg.Value.Encode()
- require.NoError(m.t, err)
- j, err := insaneJSON.DecodeBytes(val)
- if err != nil {
- m.t.Fatalf("decoding json: %v", err)
- }
- topic := j.Dig(topicField).AsString()
- if msg.Topic != topic && (topic == "" && msg.Topic != defaultTopic) {
- m.t.Fatalf("wrong topic: %s, expecting: %s for message: %s", msg.Topic, topic, string(val))
- }
-}
-
-func (m *mockProducer) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) {
- m.ensureTopic(msg)
- return -1, -1, nil
-}
-
-func (m *mockProducer) SendMessages(msgs []*sarama.ProducerMessage) error {
- for _, msg := range msgs {
- m.ensureTopic(msg)
- }
- return nil
-}
-
-func (m *mockProducer) Close() error {
- return nil
}
func newEvent(t *testing.T, topicField, topicVal, key, val string) *pipeline.Event {
@@ -113,12 +64,12 @@ func FuzzKafka(f *testing.F) {
config: &config,
avgEventSize: 10,
controller: nil,
- producer: nil,
+ client: nil,
batcher: nil,
}
f.Fuzz(func(t *testing.T, topicField, topicVal, key, val string) {
- p.producer = &mockProducer{
+ p.client = &mockProducer{
t: t,
}