fixup connected status

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

kgo.go

@@ -142,9 +142,10 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, *ho
 			}
 			return nil, nil, err
 		}
-	}
+		k.connected.Store(1)
 		return c, htracer, nil
 	}
+}
 
 func (k *Broker) Connect(ctx context.Context) error {
 	if k.connected.Load() == 1 {
@@ -404,6 +405,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,