kgo.go

@@ -568,7 +568,8 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
 		messagePool:  messagePool,
 	}
 
-	kopts := append(b.kopts,
+	kopts := append(
+		[]kgo.Opt{
 			kgo.ConsumerGroup(options.Group),
 			kgo.ConsumeTopics(topic),
 			kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
@@ -581,6 +582,8 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
 			kgo.AutoCommitCallback(sub.autocommit),
 			kgo.AutoCommitMarks(),
 			kgo.WithHooks(sub),
+		},
+		b.kopts...,
 	)
 
 	if options.Context != nil {