fix double init

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

kgo.go

@@ -53,6 +53,7 @@ var DefaultRetryBackoffFn = func() func(int) time.Duration {
 }()
 
 type Broker struct {
+	init      bool
 	c         *kgo.Client
 	kopts     []kgo.Opt
 	connected bool
@@ -155,6 +156,10 @@ func (k *Broker) Init(opts ...broker.Option) error {
 	k.Lock()
 	defer k.Unlock()
 
+	if len(opts) == 0 && k.init {
+		return nil
+	}
+
 	for _, o := range opts {
 		o(&k.opts)
 	}
@@ -178,6 +183,7 @@ func (k *Broker) Init(opts ...broker.Option) error {
 		}
 	}
 
+	k.init = true
 	return nil
 }