fix: delay start-up until Kafka is ready

runner/sidecar/handler.go

@@ -5,8 +5,13 @@ import (
 	"github.com/Shopify/sarama"
 )
 
+func newHandler(f func(ctx context.Context, msg []byte) error) *handler {
+	return &handler{f: f}
+}
+
 type handler struct {
-	f  func(context.Context, []byte) error
+	f       func(context.Context, []byte) error
+	started bool
 }
 
 func (h *handler) Setup(sarama.ConsumerGroupSession) error {
@@ -22,6 +27,7 @@ func (h *handler) Close() error {
 }
 
 func (h *handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	h.started = true
 	for m := range claim.Messages() {
 		msg := m.Value
 		if err := h.f(context.Background(), msg); err != nil {

runner/sidecar/sources.go

@@ -137,7 +137,7 @@ func connectKafkaSource(ctx context.Context, x *dfv1.Kafka, sourceName string, f
 		logger.Info("closing kafka consumer group", "source", sourceName)
 		return group.Close()
 	})
-	handler := &handler{f: f}
+	handler := newHandler(f)
 	go wait.JitterUntil(func() {
 		if err := group.Consume(ctx, []string{x.Topic}, handler); err != nil {
 			logger.Error(err, "failed to create kafka consumer")
@@ -147,6 +147,9 @@ func connectKafkaSource(ctx context.Context, x *dfv1.Kafka, sourceName string, f
 		logger.Info("closing kafka handler", "source", sourceName)
 		return handler.Close()
 	})
+	for ; !handler.started; {
+		time.Sleep(time.Second)
+	}
 	if leadReplica() {
 		registerKafkaSetPendingHook(x, sourceName, client, config, groupName)
 	}