.golangci.yml

@@ -1,5 +0,0 @@
-run:
-  concurrency: 8
-  timeout: 5m
-  issues-exit-code: 1
-  tests: true

go.mod

@@ -1,15 +1,15 @@
 module go.unistack.org/micro-broker-kgo/v4
 
-go 1.24.0
+go 1.25
 
 require (
 	github.com/stretchr/testify v1.11.1
-	github.com/twmb/franz-go v1.20.2
+	github.com/twmb/franz-go v1.20.5
 	github.com/twmb/franz-go/pkg/kadm v1.17.1
 	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0
 	go.opentelemetry.io/otel v1.38.0
-	go.unistack.org/micro/v4 v4.1.24
+	go.unistack.org/micro/v4 v4.1.25
 )
 
 require (
@@ -21,7 +21,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/spf13/cast v1.10.0 // indirect
 	go.unistack.org/micro-proto/v4 v4.1.0 // indirect
-	golang.org/x/crypto v0.43.0 // indirect
+	golang.org/x/crypto v0.46.0 // indirect
 	google.golang.org/protobuf v1.36.10 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -6,8 +6,6 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
-github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=
 github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -20,36 +18,28 @@ github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU
 github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
+github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
-github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
+github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY=
 github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
-github.com/twmb/franz-go v1.19.5 h1:W7+o8D0RsQsedqib71OVlLeZ0zI6CbFra7yTYhZTs5Y=
+github.com/twmb/franz-go v1.20.5 h1:Gj9jdkvlddf8pdrehvtDHLPult5JS8q65oITUff6dXo=
-github.com/twmb/franz-go v1.19.5/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM=
+github.com/twmb/franz-go v1.20.5/go.mod h1:gZmp2nTNfKuiKKND8qAsv28VdMlr/Gf4BIcsj99Bmtk=
-github.com/twmb/franz-go v1.20.2 h1:CiwhyKZHW6vqSHJkh+RTxFAJkio0jBjM/JQhx/HZ72A=
-github.com/twmb/franz-go v1.20.2/go.mod h1:YCnepDd4gl6vdzG03I5Wa57RnCTIC6DVEyMpDX/J8UA=
-github.com/twmb/franz-go/pkg/kadm v1.16.1 h1:IEkrhTljgLHJ0/hT/InhXGjPdmWfFvxp7o/MR7vJ8cw=
-github.com/twmb/franz-go/pkg/kadm v1.16.1/go.mod h1:Ue/ye1cc9ipsQFg7udFbbGiFNzQMqiH73fGC2y0rwyc=
 github.com/twmb/franz-go/pkg/kadm v1.17.1 h1:Bt02Y/RLgnFO2NP2HVP1kd2TFtGRiJZx+fSArjZDtpw=
 github.com/twmb/franz-go/pkg/kadm v1.17.1/go.mod h1:s4duQmrDbloVW9QTMXhs6mViTepze7JLG43xwPcAeTg=
 github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3 h1:p24opKWPySAy8xSl8NqRgOv7Q+bX7kdrQirBVRJzQfo=
 github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3/go.mod h1:7uQs3Ae6HkWT1Y9elMbqtAcNFCI0y6+iS+Phw49L49U=
-github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg=
-github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE=
 github.com/twmb/franz-go/pkg/kmsg v1.12.0 h1:CbatD7ers1KzDNgJqPbKOq0Bz/WLBdsTH75wgzeVaPc=
 github.com/twmb/franz-go/pkg/kmsg v1.12.0/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY=
 go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8=
 go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM=
 go.unistack.org/micro-proto/v4 v4.1.0 h1:qPwL2n/oqh9RE3RTTDgt28XK3QzV597VugQPaw9lKUk=
 go.unistack.org/micro-proto/v4 v4.1.0/go.mod h1:ArmK7o+uFvxSY3dbJhKBBX4Pm1rhWdLEFf3LxBrMtec=
-go.unistack.org/micro/v4 v4.1.21 h1:F9PrbI1BhXSDS0FopwcO5wWrT2Xh38w9VhVZZBHjfg8=
+go.unistack.org/micro/v4 v4.1.25 h1:JznyZN4K4HrfaEJNPC8z86EBMqC57NiQnmTrNR/Lu6M=
-go.unistack.org/micro/v4 v4.1.21/go.mod h1:nlBXTbx0rQrSZX4HPp2m57PHmpuGPWUd0O+jpUIiPto=
+go.unistack.org/micro/v4 v4.1.25/go.mod h1:PK9s7QW65HrWf26FWkdmPEQoYhsdV0e1WwThwspjpls=
-go.unistack.org/micro/v4 v4.1.24 h1:PbkSWJS3ssB5A0y0tOdOw6u9e2Mk4yombF4yR0Jshvo=
+golang.org/x/crypto v0.46.0 h1:cKRW/pmt1pKAfetfu+RCEvjvZkA9RimPbh7bhFjGVBU=
-go.unistack.org/micro/v4 v4.1.24/go.mod h1:nlBXTbx0rQrSZX4HPp2m57PHmpuGPWUd0O+jpUIiPto=
+golang.org/x/crypto v0.46.0/go.mod h1:Evb/oLKmMraqjZ2iQTwDwvCtJkczlDuTmdJXoZVzqU0=
-golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04=
-golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0=
 google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
 google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

kgo_test.go

@@ -48,6 +48,7 @@ func TestFail(t *testing.T) {
 			kg.ClientID("test"),
 			kg.FetchMaxBytes(10*1024*1024),
 			kg.AllowAutoTopicCreation(),
+			kg.MaxBufferedRecords(10),
 		),
 	)
 
@@ -231,3 +232,161 @@ func TestPubSub(t *testing.T) {
 	}()
 	<-done
 }
+
+func TestKillConsumers_E2E_Rebalance(t *testing.T) {
+	logger.DefaultLogger = slog.NewLogger()
+	if err := logger.DefaultLogger.Init(logger.WithLevel(logger.InfoLevel)); err != nil {
+		t.Fatal(err)
+	}
+
+	bLogger := broker.Logger(logger.DefaultLogger.Clone(logger.WithLevel(logger.DebugLevel)))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	b1 := kgo.NewBroker(
+		broker.ContentType("application/octet-stream"),
+		broker.Codec("application/octet-stream", codec.NewCodec()),
+		broker.Addrs(cluster.ListenAddrs()...),
+		bLogger,
+		kgo.CommitInterval(500*time.Millisecond),
+		kgo.Options(
+			kg.ClientID("test-1"),
+			kg.FetchMaxBytes(10*1024*1024),
+			kg.AllowAutoTopicCreation(),
+			kg.MaxBufferedRecords(10),
+		),
+	)
+	if err := b1.Init(); err != nil {
+		t.Fatal(err)
+	}
+	if err := b1.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = b1.Disconnect(context.Background()) }()
+
+	b2 := kgo.NewBroker(
+		broker.ContentType("application/octet-stream"),
+		broker.Codec("application/octet-stream", codec.NewCodec()),
+		broker.Addrs(cluster.ListenAddrs()...),
+		bLogger,
+		kgo.CommitInterval(500*time.Millisecond),
+		kgo.Options(
+			kg.ClientID("test-2"),
+			kg.FetchMaxBytes(10*1024*1024),
+			kg.AllowAutoTopicCreation(),
+			kg.MaxBufferedRecords(10),
+		),
+	)
+	if err := b2.Init(); err != nil {
+		t.Fatal(err)
+	}
+	if err := b2.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = b2.Disconnect(context.Background()) }()
+
+	const topic = "test.kill"
+	const total = int64(1000)
+
+	var (
+		processed int64
+		c1Count   int64
+		c2Count   int64
+	)
+
+	done := make(chan struct{})
+
+	h1 := func(msg broker.Message) error {
+		time.Sleep(2 * time.Millisecond)
+
+		atomic.AddInt64(&processed, 1)
+		atomic.AddInt64(&c1Count, 1)
+
+		if atomic.LoadInt64(&processed) >= total {
+			select {
+			case <-done:
+			default:
+				close(done)
+			}
+		}
+		return msg.Ack()
+	}
+
+	h2 := func(msg broker.Message) error {
+		time.Sleep(2 * time.Millisecond)
+
+		atomic.AddInt64(&processed, 1)
+		atomic.AddInt64(&c2Count, 1)
+
+		if atomic.LoadInt64(&processed) >= total {
+			select {
+			case <-done:
+			default:
+				close(done)
+			}
+		}
+		return msg.Ack()
+	}
+
+	sub1, err := b1.Subscribe(ctx, topic, h1,
+		broker.SubscribeAutoAck(true),
+		broker.SubscribeGroup(group),
+		broker.SubscribeBodyOnly(true),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub1.Unsubscribe(context.Background()) }()
+
+	go func() {
+		for atomic.LoadInt64(&processed) < total {
+			batchSize := int64(10)
+			msgs := make([]broker.Message, 0, batchSize)
+			for i := int64(0); i < batchSize; i++ {
+				m, _ := b1.NewMessage(ctx, metadata.New(0), []byte("msg"))
+				msgs = append(msgs, m)
+			}
+			_ = b1.Publish(ctx, topic, msgs...)
+
+			time.Sleep(5 * time.Millisecond)
+		}
+	}()
+
+	time.Sleep(200 * time.Millisecond)
+
+	// второй consumer подключается -> KAFKA REBALANCE
+	sub2, err := b2.Subscribe(ctx, topic, h2,
+		broker.SubscribeAutoAck(true),
+		broker.SubscribeGroup(group),
+		broker.SubscribeBodyOnly(true),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub2.Unsubscribe(context.Background()) }()
+
+	// ждём окончания
+	select {
+	case <-done:
+		t.Log("DONE")
+	case <-ctx.Done():
+		t.Fatalf("timeout: processed=%d of %d (c1=%d, c2=%d)",
+			atomic.LoadInt64(&processed),
+			total,
+			atomic.LoadInt64(&c1Count),
+			atomic.LoadInt64(&c2Count),
+		)
+	}
+
+	if got := atomic.LoadInt64(&processed); got != total {
+		t.Fatalf("processed %d, want %d", got, total)
+	}
+
+	if atomic.LoadInt64(&c1Count) == 0 {
+		t.Fatalf("consumer1 did not process any messages")
+	}
+	if atomic.LoadInt64(&c2Count) == 0 {
+		t.Fatalf("consumer2 did not process any messages (rebalance/killConsumers likely broken)")
+	}
+}

subscriber.go

@@ -79,8 +79,10 @@ func (s *Subscriber) Unsubscribe(ctx context.Context) error {
 		kc[ctp.t] = append(kc[ctp.t], ctp.p)
 	}
 	s.killConsumers(ctx, kc)
+	s.mu.Lock()
 	close(s.done)
 	s.closed = true
+	s.mu.Unlock()
 	s.c.ResumeFetchTopics(s.topic)
 
 	return nil
@@ -119,9 +121,9 @@ func (s *Subscriber) poll(ctx context.Context) {
 			}
 			fetches.EachError(func(t string, p int32, err error) {
 				tps := tp{t, p}
-				s.mu.Lock()
+				s.mu.RLock()
 				c := s.consumers[tps]
-				s.mu.Unlock()
+				s.mu.RUnlock()
 				if c != nil {
 					c.recs <- newErrorFetchTopicPartition(err, t, p)
 				}
@@ -129,9 +131,9 @@ func (s *Subscriber) poll(ctx context.Context) {
 
 			fetches.EachPartition(func(p kgo.FetchTopicPartition) {
 				tps := tp{p.Topic, p.Partition}
-				s.mu.Lock()
+				s.mu.RLock()
 				c := s.consumers[tps]
-				s.mu.Unlock()
+				s.mu.RUnlock()
 				if c != nil {
 					c.recs <- p
 				}
@@ -150,19 +152,38 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
 			tps := tp{topic, partition}
 			s.mu.Lock()
 			pc, ok := s.consumers[tps]
+			if ok {
+				delete(s.consumers, tps)
+			}
 			s.mu.Unlock()
 			if !ok || pc == nil {
 				continue
 			}
-			s.mu.Lock()
+
-			delete(s.consumers, tps)
-			s.mu.Unlock()
-			close(pc.quit)
 			if s.kopts.Logger.V(logger.DebugLevel) {
-				s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] waiting for work to finish topic %s partition %d", topic, partition))
+				s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] killing consumer topic %s partition %d", topic, partition))
 			}
+
+			close(pc.quit)
+
 			wg.Add(1)
-			go func() { <-pc.done; wg.Done() }()
+			go func(c *consumer, t string, p int32) {
+				defer wg.Done()
+
+				timeout := time.NewTimer(s.kopts.GracefulTimeout)
+				defer timeout.Stop()
+
+				select {
+				case <-c.done:
+					if s.kopts.Logger.V(logger.DebugLevel) {
+						s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] consumer stopped topic %s partition %d", t, p))
+					}
+				case <-timeout.C:
+					if s.kopts.Logger.V(logger.DebugLevel) {
+						s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] timeout waiting for consumer topic %s partition %d", t, p))
+					}
+				}
+			}(pc, topic, partition)
 		}
 	}
 }
@@ -197,7 +218,6 @@ func (s *Subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[str
 	if s.kopts.Logger.V(logger.DebugLevel) {
 		s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] revoked %#+v", revoked))
 	}
-	s.killConsumers(ctx, revoked)
 	if err := c.CommitMarkedOffsets(ctx); err != nil {
 		s.mu.Lock()
 		tpc := make(map[tp]*consumer, len(s.consumers))
@@ -209,6 +229,7 @@ func (s *Subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[str
 			}
 		}
 	}
+	s.killConsumers(ctx, revoked)
 }
 
 func (s *Subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[string][]int32) {
@@ -251,7 +272,10 @@ func (pc *consumer) consume() {
 		select {
 		case <-pc.quit:
 			return
-		case p := <-pc.recs:
+		case p, ok := <-pc.recs:
+			if !ok {
+				return
+			}
 			if p.Err != nil || p.FetchPartition.Err != nil {
 
 				if p.Err != nil {
@@ -274,6 +298,11 @@ func (pc *consumer) consume() {
 			}
 
 			for _, record := range p.Records {
+				select {
+				case <-pc.quit:
+					return
+				default:
+				}
 				ctx, sp := pc.htracer.WithProcessSpan(record)
 				ts := time.Now()
 				pc.kopts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", record.Topic, "topic", record.Topic).Inc()
@@ -300,12 +329,25 @@ func (pc *consumer) consume() {
 				pm.hdr.Set("Micro-Key", string(record.Key))
 				pm.hdr.Set("Micro-Timestamp", strconv.FormatInt(record.Timestamp.Unix(), 10))
 
-				switch h := pc.handler.(type) {
+				processCtx, cancel := context.WithTimeout(ctx, pc.kopts.GracefulTimeout)
-				case func(broker.Message) error:
+				errChan := make(chan error, 1)
-					err = h(pm)
+
-				case func([]broker.Message) error:
+				go func() {
-					err = h([]broker.Message{pm})
+					switch h := pc.handler.(type) {
+					case func(broker.Message) error:
+						errChan <- h(pm)
+					case func([]broker.Message) error:
+						errChan <- h([]broker.Message{pm})
+					}
+				}()
+
+				select {
+				case err = <-errChan:
+				case <-processCtx.Done():
+					//err = fmt.Errorf("[kgo] message processing timeout topic %s partition %d offset %d", record.Topic, record.Partition, record.Offset)
+					err = processCtx.Err()
 				}
+				cancel()
 
 				pc.kopts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", record.Topic, "topic", record.Topic).Dec()
 				if err != nil {
@@ -335,6 +377,8 @@ func (pc *consumer) consume() {
 					continue
 				}
 
+				pc.kopts.Logger.Debug(pc.kopts.Context, fmt.Sprintf("[kgo] message not acknowledged topic %s partition %d offset %d", record.Topic, record.Partition, record.Offset))
+
 				pm := pc.newErrorMessage(ErrLostMessage, p.Topic, p.Partition)
 				switch h := pc.handler.(type) {
 				case func(broker.Message) error: