disable connected state changes now

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

broker.go

@@ -27,7 +27,7 @@ var (
 
 func (m *hookEvent) OnGroupManageError(err error) {
 	if err != nil {
-		m.connected.Store(0)
+		// m.connected.Store(0)
 		if m.fatalOnError {
 			m.log.Fatal(context.TODO(), "kgo.OnGroupManageError", err)
 		}
@@ -36,7 +36,7 @@ func (m *hookEvent) OnGroupManageError(err error) {
 
 func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
 	if err != nil {
-		m.connected.Store(0)
+		// m.connected.Store(0)
 		if m.fatalOnError {
 			m.log.Fatal(context.TODO(), "kgo.OnBrokerConnect", err)
 		}
@@ -44,12 +44,12 @@ func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net
 }
 
 func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) {
-	m.connected.Store(0)
+	// m.connected.Store(0)
 }
 
 func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
 	if err != nil {
-		m.connected.Store(0)
+		// m.connected.Store(0)
 		if m.fatalOnError {
 			m.log.Fatal(context.TODO(), "kgo.OnBrokerWrite", err)
 		}
@@ -58,12 +58,12 @@ func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.D
 
 func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
 	if err != nil {
-		m.connected.Store(0)
+		// m.connected.Store(0)
 	}
 }
 
 func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
 	if err != nil {
-		m.connected.Store(0)
+		// m.connected.Store(0)
 	}
 }

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 {
@@ -267,7 +268,7 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
 		k.connected.Store(1)
 	}
 	k.Unlock()
-	fmt.Printf("EEE\n")
+
 	options := broker.NewPublishOptions(opts...)
 	records := make([]*kgo.Record, 0, len(msgs))
 	var errs []string
@@ -322,9 +323,9 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
 		return nil
 	}
 	ts := time.Now()
-	fmt.Printf("SSSSSSEEE\n")
+
 	results := k.c.ProduceSync(ctx, records...)
-	fmt.Printf("SSSSSS\n")
+
 	te := time.Since(ts)
 	for _, result := range results {
 		k.opts.Meter.Summary(semconv.PublishMessageLatencyMicroseconds, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Update(te.Seconds())
@@ -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,

kgo_test.go

@@ -2,7 +2,6 @@ package kgo_test
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"strings"
 	"sync/atomic"
@@ -214,7 +213,7 @@ func TestPubSub(t *testing.T) {
 				if prc := atomic.LoadInt64(&idx); prc == msgcnt {
 					close(done)
 				} else {
-					fmt.Printf("processed %v\n", prc)
+					t.Logf("processed %v\n", prc)
 				}
 			case <-ticker.C:
 				close(done)

subscriber.go

@@ -160,7 +160,10 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
 	for topic, partitions := range lost {
 		for _, partition := range partitions {
 			tps := tp{topic, partition}
-			pc := s.consumers[tps]
+			pc, ok := s.consumers[tps]
+			if !ok {
+				continue
+			}
 			delete(s.consumers, tps)
 			close(pc.quit)
 			if s.kopts.Logger.V(logger.DebugLevel) {