diff --git a/broker/segmentio/broker_test.go b/broker/segmentio/broker_test.go
new file mode 100644
index 0000000..8406cf7
--- /dev/null
+++ b/broker/segmentio/broker_test.go
@@ -0,0 +1,444 @@
+package segmentio_test
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/segmentio/kafka-go"
+	segmentio "github.com/unistack-org/micro-broker-segmentio/v3"
+	victoriameter "github.com/unistack-org/micro-meter-victoriametrics/v3"
+	https "github.com/unistack-org/micro-server-http/v3"
+	"github.com/unistack-org/micro/v3/broker"
+	"github.com/unistack-org/micro/v3/codec"
+	"github.com/unistack-org/micro/v3/logger"
+	"github.com/unistack-org/micro/v3/meter"
+	meterhandler "github.com/unistack-org/micro/v3/meter/handler"
+	"github.com/unistack-org/micro/v3/server"
+)
+
+type lg struct{}
+
+func (l *lg) Printf(format string, args ...interface{}) {
+	//	logger.Infof(context.Background(), format, args...)
+}
+
+var (
+	bm = &broker.Message{
+		Header: map[string]string{"hkey": "hval"},
+		Body:   []byte(`"body"`),
+	}
+)
+
+func TestSub(t *testing.T) {
+	topic := fmt.Sprintf("test_topic")
+	if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
+		t.Skip()
+	}
+
+	logger.DefaultLogger.Init(logger.WithLevel(logger.ErrorLevel))
+	ctx := context.Background()
+
+	var addrs []string
+	if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
+		addrs = []string{"127.0.0.1:9092"}
+	} else {
+		addrs = strings.Split(addr, ",")
+	}
+
+	meter.DefaultMeter = victoriameter.NewMeter()
+
+	s := https.NewServer(server.Context(ctx), server.Address("127.0.0.1:0"), server.Codec("text/plain", codec.NewCodec()))
+	if err := s.Init(); err != nil {
+		t.Fatal(err)
+	}
+	if err := meterhandler.RegisterMeterServer(s, meterhandler.NewHandler()); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Start(); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := s.Stop(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	segmentio.DefaultWriterConfig.Async = true
+	segmentio.DefaultWriterConfig.BatchTimeout = 1 * time.Second
+	segmentio.DefaultWriterConfig.RequiredAcks = int(kafka.RequireAll)
+	segmentio.DefaultReaderConfig.StartOffset = kafka.FirstOffset
+	segmentio.DefaultReaderConfig.MinBytes = 1024 * 10        // 10 kb
+	segmentio.DefaultReaderConfig.MaxBytes = 1024 * 1024 * 20 // 20 Mb
+	segmentio.DefaultReaderConfig.MaxWait = 20 * time.Second  // 20s
+	segmentio.DefaultReaderConfig.QueueCapacity = 500
+	segmentio.DefaultReaderConfig.ReadBackoffMin = 2 * time.Second
+	segmentio.DefaultReaderConfig.ReadBackoffMax = 5 * time.Second
+	segmentio.DefaultReaderConfig.Logger = &lg{}
+	segmentio.DefaultReaderConfig.CommitInterval = 1 * time.Second
+	brk := segmentio.NewBroker(broker.Context(ctx), broker.Addrs(addrs...), segmentio.StatsInterval(5*time.Second),
+		segmentio.ClientID("test_sub"),
+	)
+	t.Logf("init")
+	if err := brk.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	t.Logf("connect")
+	if err := brk.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		t.Logf("disconnect")
+		if err := brk.Disconnect(ctx); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	fmt.Printf("prefill topic\n")
+	go func() {
+		for i := 0; i < 900000; i++ {
+			//	brk.Publish(ctx, topic, bm)
+			time.Sleep(1 * time.Second)
+		}
+	}()
+	fmt.Printf("prefill complete\n")
+
+	var cnt uint64
+	var wait atomic.Value
+	wait.Store(true)
+
+	done := make(chan struct{})
+	fn := func(msg broker.Event) error {
+		if wait.Load().(bool) {
+			wait.Store(false)
+			fmt.Printf("done ready\n")
+			close(done)
+		}
+		atomic.AddUint64(&cnt, 1)
+		return msg.Ack()
+	}
+
+	sub, err := brk.Subscribe(ctx, topic, fn, broker.SubscribeGroup("test"), broker.SubscribeBodyOnly(true))
+	if err != nil {
+		t.Fatal(err)
+	}
+	fmt.Printf("wait for ready\n")
+	<-done
+	fmt.Printf("wait for bench\n")
+	fmt.Printf("start %s\n", time.Now().String())
+	<-time.After(20 * time.Second)
+	fmt.Printf("stop %s\n", time.Now().String())
+	rcnt := atomic.LoadUint64(&cnt)
+
+	req, err := http.NewRequest(http.MethodGet, "http://"+s.Options().Address+"/metrics", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Add("Content-Type", "text/plain")
+
+	rsp, err := (&http.Client{}).Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer rsp.Body.Close()
+
+	buf, err := io.ReadAll(rsp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fmt.Printf("unsub\n")
+	if err := sub.Unsubscribe(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	t.Logf("metrics: \n%s\n", buf)
+	t.Logf("mesage count %d\n", rcnt)
+}
+
+func BenchmarkPub(b *testing.B) {
+	if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
+		b.Skip()
+	}
+
+	logger.DefaultLogger.Init(logger.WithLevel(logger.TraceLevel))
+	ctx := context.Background()
+
+	var addrs []string
+	if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
+		addrs = []string{"127.0.0.1:9092"}
+	} else {
+		addrs = strings.Split(addr, ",")
+	}
+
+	meter.DefaultMeter = victoriameter.NewMeter()
+
+	s := https.NewServer(server.Context(ctx), server.Address("127.0.0.1:0"), server.Codec("text/plain", codec.NewCodec()))
+	if err := s.Init(); err != nil {
+		b.Fatal(err)
+	}
+	if err := meterhandler.RegisterMeterServer(s, meterhandler.NewHandler()); err != nil {
+		b.Fatal(err)
+	}
+
+	if err := s.Start(); err != nil {
+		b.Fatal(err)
+	}
+	defer func() {
+		if err := s.Stop(); err != nil {
+			b.Fatal(err)
+		}
+	}()
+
+	segmentio.DefaultWriterConfig.Async = true
+	segmentio.DefaultWriterConfig.BatchTimeout = 1 * time.Second
+	segmentio.DefaultWriterConfig.RequiredAcks = int(kafka.RequireAll)
+	fn := func(msgs []kafka.Message, err error) {
+		if err != nil {
+			b.Logf("err %v", err)
+		}
+	}
+	brk := segmentio.NewBroker(broker.Context(ctx), broker.Addrs(addrs...), segmentio.StatsInterval(1*time.Second),
+		segmentio.WriterCompletionFunc(fn))
+	b.Logf("init")
+	if err := brk.Init(); err != nil {
+		b.Fatal(err)
+	}
+
+	b.Logf("connect")
+	if err := brk.Connect(ctx); err != nil {
+		b.Fatal(err)
+	}
+	defer func() {
+		b.Logf("disconnect")
+		if err := brk.Disconnect(ctx); err != nil {
+			b.Fatal(err)
+		}
+	}()
+
+	cnt := 0
+	b.ResetTimer()
+	for n := 0; n < b.N; n++ {
+		if err := brk.Publish(ctx, "test_topic", bm); err != nil {
+			b.Fatal(err)
+		}
+		cnt++
+	}
+
+	req, err := http.NewRequest(http.MethodGet, "http://"+s.Options().Address+"/metrics", nil)
+	if err != nil {
+		b.Fatal(err)
+	}
+	req.Header.Add("Content-Type", "text/plain")
+
+	rsp, err := (&http.Client{}).Do(req)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	defer rsp.Body.Close()
+
+	buf, err := io.ReadAll(rsp.Body)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	b.Logf("metrics: \n%s\n", buf)
+	b.Logf("mesage count %d\n", cnt)
+}
+
+func BenchmarkPubSub(b *testing.B) {
+	b.Skip()
+	ctx := context.Background()
+	topic := fmt.Sprintf("test_topic")
+	var addrs []string
+	if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
+		addrs = []string{"127.0.0.1:9092"}
+	} else {
+		addrs = strings.Split(addr, ",")
+	}
+
+	segmentio.DefaultWriterConfig.Async = true
+	segmentio.DefaultWriterConfig.BatchTimeout = 1 * time.Second
+	segmentio.DefaultReaderConfig.CommitInterval = 2 * time.Second
+	brk := segmentio.NewBroker(broker.Context(ctx), broker.Addrs(addrs...), segmentio.StatsInterval(1*time.Minute))
+	if err := brk.Init(); err != nil {
+		b.Fatal(err)
+	}
+
+	if err := brk.Connect(ctx); err != nil {
+		b.Fatal(err)
+	}
+	defer func() {
+		if err := brk.Disconnect(ctx); err != nil {
+			b.Fatal(err)
+		}
+	}()
+
+	wait := true
+	var cnt uint64
+	fn := func(msg broker.Event) error {
+		if wait {
+			wait = false
+		}
+		atomic.AddUint64(&cnt, 1)
+		return msg.Ack()
+	}
+
+	if err := brk.Publish(ctx, topic, bm); err != nil {
+		b.Fatal(err)
+	}
+
+	sub, err := brk.Subscribe(ctx, topic, fn, broker.SubscribeGroup("test"), broker.SubscribeBodyOnly(true))
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer func() {
+		if err := sub.Unsubscribe(ctx); err != nil {
+			b.Fatal(err)
+		}
+	}()
+
+	for {
+		if !wait {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+	b.ResetTimer()
+	var result error
+	sent := uint64(0)
+	for n := 0; n < b.N; n++ {
+		if err := brk.Publish(ctx, topic, bm); err != nil {
+			b.Fatal(err)
+		} else {
+			result = err
+		}
+		sent++
+	}
+
+	b.Logf("publish done")
+	for {
+		c := atomic.LoadUint64(&cnt)
+		if c >= sent {
+			break
+		}
+		fmt.Printf("c %d seen %d\n", c, sent)
+		time.Sleep(1 * time.Second)
+	}
+	_ = result
+	fmt.Printf("c %d seen %d\n", atomic.LoadUint64(&cnt), sent)
+}
+
+func TestPubSub(t *testing.T) {
+	if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
+		t.Skip()
+	}
+
+	logger.DefaultLogger.Init(logger.WithLevel(logger.ErrorLevel))
+	ctx := context.Background()
+
+	var addrs []string
+	if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
+		addrs = []string{"127.0.0.1:9092"}
+	} else {
+		addrs = strings.Split(addr, ",")
+	}
+
+	meter.DefaultMeter = victoriameter.NewMeter()
+
+	s := https.NewServer(server.Context(ctx), server.Address("127.0.0.1:0"), server.Codec("text/plain", codec.NewCodec()))
+	if err := s.Init(); err != nil {
+		t.Fatal(err)
+	}
+	if err := meterhandler.RegisterMeterServer(s, meterhandler.NewHandler()); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Start(); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := s.Stop(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	b := segmentio.NewBroker(broker.Context(ctx), broker.Addrs(addrs...), segmentio.StatsInterval(500*time.Millisecond),
+		segmentio.ClientID("test_pubsub"))
+	t.Logf("init")
+	if err := b.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	t.Logf("connect")
+	if err := b.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		t.Logf("disconnect")
+		if err := b.Disconnect(ctx); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	wait := true
+	fn := func(msg broker.Event) error {
+		wait = false
+		return msg.Ack()
+	}
+
+	t.Logf("subscribe")
+	sub, err := b.Subscribe(ctx, "test_topic", fn, broker.SubscribeGroup("test"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		t.Logf("unsubscribe")
+		if err := sub.Unsubscribe(ctx); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	if err := b.Publish(ctx, "test_topic", bm); err != nil {
+		t.Fatal(err)
+	}
+
+	for {
+		if !wait {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+
+	req, err := http.NewRequest(http.MethodGet, "http://"+s.Options().Address+"/metrics", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Add("Content-Type", "text/plain")
+
+	rsp, err := (&http.Client{}).Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer rsp.Body.Close()
+
+	buf, err := io.ReadAll(rsp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Logf("metrics: \n%s\n", buf)
+}
diff --git a/go.mod b/go.mod
index 93b38c2..b0d8e54 100644
--- a/go.mod
+++ b/go.mod
@@ -3,16 +3,17 @@ module github.com/unistack-org/micro-tests
 go 1.16
 
 require (
-	github.com/frankban/quicktest v1.11.3 // indirect
 	github.com/google/uuid v1.3.0
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/prometheus/client_golang v1.10.0
 	github.com/prometheus/client_model v0.2.0
+	github.com/segmentio/kafka-go v0.4.17
 	github.com/stretchr/testify v1.7.0
 	github.com/unistack-org/micro-api-handler-rpc/v3 v3.3.0
 	github.com/unistack-org/micro-api-router-register/v3 v3.2.2
 	github.com/unistack-org/micro-api-router-static/v3 v3.2.1
 	github.com/unistack-org/micro-broker-http/v3 v3.3.1
+	github.com/unistack-org/micro-broker-segmentio/v3 v3.4.1
 	//github.com/unistack-org/micro-client-drpc/v3 v3.0.0-00010101000000-000000000000
 	github.com/unistack-org/micro-client-grpc/v3 v3.3.3
 	github.com/unistack-org/micro-client-http/v3 v3.4.5
@@ -34,8 +35,9 @@ require (
 	github.com/unistack-org/micro-server-tcp/v3 v3.3.2
 	github.com/unistack-org/micro-wrapper-trace-opentracing/v3 v3.2.0
 	github.com/unistack-org/micro/v3 v3.4.9
+	golang.org/x/net v0.0.0-20210716203947-853a461950ff // indirect
 	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
-	google.golang.org/genproto v0.0.0-20210716133855-ce7ef5c701ea
+	google.golang.org/genproto v0.0.0-20210719143636-1d5a45f8e492
 	google.golang.org/grpc v1.39.0
 	google.golang.org/protobuf v1.27.1
 	storj.io/drpc v0.0.24
@@ -48,7 +50,5 @@ require (
 //replace github.com/unistack-org/micro-client-http/v3 => ../micro-client-http
 //replace github.com/unistack-org/micro-client-drpc/v3 => ../micro-client-drpc
 //replace github.com/unistack-org/micro-broker-segmentio/v3 => ../micro-broker-segmentio
-
 //replace github.com/unistack-org/micro/v3 => ../micro
-
 //replace github.com/unistack-org/micro-proto => ../micro-proto
diff --git a/go.sum b/go.sum
index 8c3fd7a..d1eb1ce 100644
--- a/go.sum
+++ b/go.sum
@@ -119,6 +119,7 @@ github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5Xh
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
+github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
 github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
@@ -202,8 +203,9 @@ github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
-github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA=
 github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -328,6 +330,9 @@ github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
+github.com/klauspost/compress v1.13.1 h1:wXr2uRxZTJXHLly6qhJabee5JqIhTRoLBhDOA74hDEQ=
+github.com/klauspost/compress v1.13.1/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
 github.com/klauspost/cpuid v1.2.3 h1:CCtW0xUnWGVINKvE/WWOYKdsPV6mawAtvQuSl8guwQs=
 github.com/klauspost/cpuid v1.2.3/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/cpuid/v2 v2.0.5 h1:qnfhwbFriwDIX51QncuNU5mEMf+6KE3t7O8V2KQl3Dg=
@@ -452,6 +457,7 @@ github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pierrec/lz4 v2.5.2+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4 v2.6.0+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM=
 github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -515,6 +521,8 @@ github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdh
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/segmentio/encoding v0.2.17 h1:cgfmPc44u1po1lz5bSgF00gLCROBjDNc7h+H7I20zpc=
 github.com/segmentio/encoding v0.2.17/go.mod h1:7E68jTSWMnNoYhHi1JbLd7NBSB6XfE4vzqhR88hDBQc=
+github.com/segmentio/kafka-go v0.4.17 h1:IyqRstL9KUTDb3kyGPOOa5VffokKWSEzN6geJ92dSDY=
+github.com/segmentio/kafka-go v0.4.17/go.mod h1:19+Eg7KwrNKy/PFhiIthEPkO8k+ac7/ZYXwYM9Df10w=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/silas/dag v0.0.0-20210121180416-41cf55125c34 h1:vBfVmA5mZhsQa2jr1FOL9nfA37N/jnbBmi5XUfviVTI=
 github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
@@ -559,6 +567,8 @@ github.com/unistack-org/micro-api-router-static/v3 v3.2.1 h1:FMcEb8NYYX72OyQuMKH
 github.com/unistack-org/micro-api-router-static/v3 v3.2.1/go.mod h1:vP8VMZrJtlimEQOgYnLJsjnaUMeULNMPBkHEK5DA7G0=
 github.com/unistack-org/micro-broker-http/v3 v3.3.1 h1:gs7TVth+rd2yOnikxstRlMIqSYas1yyxQAp9sJrhuMY=
 github.com/unistack-org/micro-broker-http/v3 v3.3.1/go.mod h1:jO77nTg/qTwT+JVQQdFENa3LzQ8sLNU1dp3sUElgog0=
+github.com/unistack-org/micro-broker-segmentio/v3 v3.4.1 h1:BH6iSZ7ztAJ2R0JXVMN7/Tlj2It13daAr1hG+/GKJrw=
+github.com/unistack-org/micro-broker-segmentio/v3 v3.4.1/go.mod h1:QbDL0VZLw/g+63Memwq23U8R+26QVZO7H8OuSjbK0sk=
 github.com/unistack-org/micro-client-grpc/v3 v3.3.3 h1:zKhxh45A4SP1b002fiFLK+vayL/YkHOpoCyZCRBmbm4=
 github.com/unistack-org/micro-client-grpc/v3 v3.3.3/go.mod h1:96DSXqLd5kSiep2brs+h7WvsZG+z3C2KO/NDaYA1Czk=
 github.com/unistack-org/micro-client-http/v3 v3.4.5 h1:nvLIMW1Bdh6R1JVSOS40Hyr6nVuHSt0yPTmA3QxKuMM=
@@ -618,6 +628,10 @@ github.com/valyala/fastrand v1.0.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002
 github.com/valyala/histogram v1.1.2 h1:vOk5VrGjMBIoPR5k6wA8vBaC8toeJ8XO0yfRjFEc1h8=
 github.com/valyala/histogram v1.1.2/go.mod h1:CZAr6gK9dbD7hYx2s8WSPh0p5x5wETjC+2b3PJVtEdg=
 github.com/vultr/govultr v0.1.4/go.mod h1:9H008Uxr/C4vFNGLqKx232C206GL0PBHzOP0809bGNA=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
+github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=
+github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.1.0/go.mod h1:5yf86TLmAcydyeJq5YvxkGPE2fm/u4myDekKRoLuqhs=
@@ -650,6 +664,7 @@ golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20190211182817-74369b46fc67/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190418165655-df01cb2cc480/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
+golang.org/x/crypto v0.0.0-20190506204251-e1dfcc566284/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -719,8 +734,9 @@ golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96b
 golang.org/x/net v0.0.0-20210415231046-e915ea6b2b7d/go.mod h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8=
 golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/net v0.0.0-20210510120150-4163338589ed/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210614182718-04defd469f4e h1:XpT3nA5TvE525Ne3hInMh6+GETgn27Zfm9dxsThnX2Q=
 golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20210716203947-853a461950ff h1:j2EK/QoxYNBsXI4R7fQkkRUk8y6wnOBI+6hgPdP/6Ds=
+golang.org/x/net v0.0.0-20210716203947-853a461950ff/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -826,7 +842,7 @@ golang.org/x/tools v0.0.0-20191216052735-49a3e744a425/go.mod h1:TB2adYChydJhpapK
 golang.org/x/tools v0.0.0-20191216173652-a0e659d51361/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200103221440-774c71fcf114/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
-golang.org/x/tools v0.1.4/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
+golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -856,8 +872,8 @@ google.golang.org/genproto v0.0.0-20191216164720-4f79533eabd1/go.mod h1:n3cpQtvx
 google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
 google.golang.org/genproto v0.0.0-20200904004341-0bd0a958aa1d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
-google.golang.org/genproto v0.0.0-20210716133855-ce7ef5c701ea h1:8ZyCcgugUqamxp/vZSEJw9CMy7VZlSWYJLLJPi/dSDA=
-google.golang.org/genproto v0.0.0-20210716133855-ce7ef5c701ea/go.mod h1:AxrInvYm1dci+enl5hChSFPOmmUF1+uAa/UsgNRWd7k=
+google.golang.org/genproto v0.0.0-20210719143636-1d5a45f8e492 h1:7yQQsvnwjfEahbNNEKcBHv3mR+HnB1ctGY/z1JXzx8M=
+google.golang.org/genproto v0.0.0-20210719143636-1d5a45f8e492/go.mod h1:ob2IJxKrgPT52GcgX759i1sleT07tiKowYBGbczaW48=
 google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
 google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=