Merge pull request 'correction create opts' (#160) from devstigneev/micro-broker-kgo:fix_opts into v4

Reviewed-on: #160

kgo.go

@@ -568,19 +568,22 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
 		messagePool:  messagePool,
 	}
 
-	kopts := append(b.kopts,
-		kgo.ConsumerGroup(options.Group),
-		kgo.ConsumeTopics(topic),
-		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
-		kgo.FetchMaxWait(1*time.Second),
-		kgo.AutoCommitInterval(commitInterval),
-		kgo.OnPartitionsAssigned(sub.assigned),
-		kgo.OnPartitionsRevoked(sub.revoked),
-		kgo.StopProducerOnDataLossDetected(),
-		kgo.OnPartitionsLost(sub.lost),
-		kgo.AutoCommitCallback(sub.autocommit),
-		kgo.AutoCommitMarks(),
-		kgo.WithHooks(sub),
+	kopts := append(
+		[]kgo.Opt{
+			kgo.ConsumerGroup(options.Group),
+			kgo.ConsumeTopics(topic),
+			kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
+			kgo.FetchMaxWait(1 * time.Second),
+			kgo.AutoCommitInterval(commitInterval),
+			kgo.OnPartitionsAssigned(sub.assigned),
+			kgo.OnPartitionsRevoked(sub.revoked),
+			kgo.StopProducerOnDataLossDetected(),
+			kgo.OnPartitionsLost(sub.lost),
+			kgo.AutoCommitCallback(sub.autocommit),
+			kgo.AutoCommitMarks(),
+			kgo.WithHooks(sub),
+		},
+		b.kopts...,
 	)
 
 	if options.Context != nil {

subscriber.go

@@ -123,7 +123,7 @@ func (s *Subscriber) poll(ctx context.Context) {
 				c := s.consumers[tps]
 				s.mu.Unlock()
 				if c != nil {
-					c.recs <- newErrorFetchTopicPartition(kgo.ErrClientClosed, t, p)
+					c.recs <- newErrorFetchTopicPartition(err, t, p)
 				}
 			})
 
@@ -199,7 +199,15 @@ func (s *Subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[str
 	}
 	s.killConsumers(ctx, revoked)
 	if err := c.CommitMarkedOffsets(ctx); err != nil {
-		s.kopts.Logger.Error(ctx, "[kgo] revoked CommitMarkedOffsets error", err)
+		s.mu.Lock()
+		tpc := make(map[tp]*consumer, len(s.consumers))
+		maps.Copy(tpc, s.consumers)
+		s.mu.Unlock()
+		for tp, c := range tpc {
+			if c != nil {
+				c.recs <- newErrorFetchTopicPartition(err, tp.t, tp.p)
+			}
+		}
 	}
 }