diff --git a/kgo.go b/kgo.go index 0e990f0..212060d 100644 --- a/kgo.go +++ b/kgo.go @@ -404,6 +404,7 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han consumers: make(map[tp]*consumer), done: make(chan struct{}), fatalOnError: fatalOnError, + connected: k.connected, } kopts := append(k.kopts,