Check for cluster connectivity in the kafka_franz input

Signed-off-by: Mihai Todor <[email protected]>

internal/impl/kafka/franz_reader_unordered.go

@@ -17,6 +17,7 @@ package kafka
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -491,6 +492,11 @@ func (f *FranzReaderUnordered) Connect(ctx context.Context) error {
 		return err
 	}
 
+	// Check connectivity to cluster
+	if err = cl.Ping(ctx); err != nil {
+		return fmt.Errorf("failed to connect to cluster: %s", err)
+	}
+
 	go func() {
 		defer func() {
 			cl.Close()