One more open broker connection

danielqsj/kafka_exporter#54 (comment) IBM/sarama#853
practo · Feb 19, 2021 · cf007c2 · cf007c2
1 parent 0e99d77
commit cf007c2
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 0 deletions.
diff --git a/controllers/redshiftsink_controller.go b/controllers/redshiftsink_controller.go
@@ -435,6 +435,8 @@ func (r *RedshiftSinkReconciler) reconcile(
 		}
 	}
 
+	klog.V(2).Infof("rsk/%v reconciling all sinkGroups", rsk.Name)
+
 	reloadDupe = sgBuilder.
 		setRedshiftSink(rsk).setClient(r.Client).setScheme(r.Scheme).
 		setType(ReloadDupeSinkGroup).

diff --git a/pkg/kafka/watcher.go b/pkg/kafka/watcher.go
@@ -162,6 +162,11 @@ func (t *kafkaWatch) consumerGroupLag(
 	}
 	offsetFetchRequest.AddPartition(topic, partition)
 
+	err = broker.Open(t.client.Config())
+	if err != nil && err != sarama.ErrAlreadyConnected {
+		return defaultLag, fmt.Errorf("Error opening broker connection again, err: %v", err)
+	}
+
 	offsetFetchResponse, err := broker.FetchOffset(&offsetFetchRequest)
 	if err != nil {
 		return defaultLag, fmt.Errorf(