remove noisy logs from event hooks and add tests (#193)

hook_event.go

@@ -0,0 +1,108 @@
+package kgo
+
+import (
+	"context"
+	"net"
+	"sync/atomic"
+	"time"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+	"go.unistack.org/micro/v4/logger"
+)
+
+type hookEvent struct {
+	log          logger.Logger
+	fatalOnError bool
+	connected    *atomic.Uint32
+}
+
+var (
+	_ kgo.HookBrokerConnect           = &hookEvent{}
+	_ kgo.HookBrokerDisconnect        = &hookEvent{}
+	_ kgo.HookBrokerRead              = &hookEvent{}
+	_ kgo.HookBrokerWrite             = &hookEvent{}
+	_ kgo.HookGroupManageError        = &hookEvent{}
+	_ kgo.HookProduceRecordUnbuffered = &hookEvent{}
+)
+
+func (m *hookEvent) OnGroupManageError(err error) {
+	switch {
+	case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
+		return
+	default:
+		ctx := context.TODO()
+		logMsg := "kgo.OnGroupManageError"
+
+		if m.fatalOnError {
+			m.log.Fatal(ctx, logMsg, err)
+		} else {
+			m.log.Error(ctx, logMsg, err)
+		}
+	}
+}
+
+func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
+	switch {
+	case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
+		return
+	default:
+		ctx := context.TODO()
+		logMsg := "kgo.OnBrokerConnect"
+
+		if m.fatalOnError {
+			m.log.Fatal(ctx, logMsg, err)
+		} else {
+			m.log.Error(ctx, logMsg, err)
+		}
+	}
+}
+
+func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) {}
+
+func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
+	switch {
+	case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
+		return
+	default:
+		ctx := context.TODO()
+		logMsg := "kgo.OnBrokerWrite"
+
+		if m.fatalOnError {
+			m.log.Fatal(ctx, logMsg, err)
+		} else {
+			m.log.Error(ctx, logMsg, err)
+		}
+	}
+}
+
+func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
+	switch {
+	case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
+		return
+	default:
+		ctx := context.TODO()
+		logMsg := "kgo.OnBrokerRead"
+
+		if m.fatalOnError {
+			m.log.Fatal(ctx, logMsg, err)
+		} else {
+			m.log.Error(ctx, logMsg, err)
+		}
+	}
+}
+
+func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
+	switch {
+	case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
+		return
+	default:
+		ctx := context.TODO()
+		logMsg := "kgo.OnProduceRecordUnbuffered"
+
+		if m.fatalOnError {
+			m.log.Fatal(ctx, logMsg, err)
+		} else {
+			m.log.Error(ctx, logMsg, err)
+		}
+	}
+}