unistack-org/micro-broker-kgo
3
0
Fork 2
Code Issues3 Pull Requests1 Actions Packages Projects Releases56 Wiki Activity

Compare commits

...

5 Commits
v3.8.56 ... v3

Author SHA1 Message Date
Vasiliy Tolstov
747e35148f disable connected state changes now 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
 2025-03-12 20:40:30 +03:00
Vasiliy Tolstov
21f06fee6c fixup panic 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
 2025-03-12 16:59:43 +03:00
Vasiliy Tolstov
f4f8793686 fixup connected status 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
 2025-03-07 11:45:36 +03:00
Vasiliy Tolstov
ec4922ad8b fixup connected status 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
 2025-03-07 11:44:56 +03:00
Vasiliy Tolstov
eaea14e5a8 fixup connected status 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
 2025-03-07 11:26:34 +03:00
3 changed files with 12 additions and 8 deletions

12
broker.go

@ -27,7 +27,7 @@ var (
func (m *hookEvent) OnGroupManageError(err error) { func (m *hookEvent) OnGroupManageError(err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
if m.fatalOnError { if m.fatalOnError {
m.log.Fatal(context.TODO(), "kgo.OnGroupManageError", err) 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) { func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
if m.fatalOnError { if m.fatalOnError {
m.log.Fatal(context.TODO(), "kgo.OnBrokerConnect", err) 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) { 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) { func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
if m.fatalOnError { if m.fatalOnError {
m.log.Fatal(context.TODO(), "kgo.OnBrokerWrite", err) 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) { func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
} }
} }
func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) { func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
} }
} }

3
kgo.go

@ -142,8 +142,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, *ho
} }
return nil, nil, err return nil, nil, err
} }
k.connected.Store(1)
return c, htracer, nil
} }
return c, htracer, nil
} }
func (k *Broker) Connect(ctx context.Context) error { func (k *Broker) Connect(ctx context.Context) error {

5
subscriber.go

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