mq/kafka: wait for a filer to register instead of panicking on an empty list (#12019)

* mq/kafka: wait for a filer to register instead of panicking on an empty list

NewSeaweedMQBrokerHandler queried the masters for filers once, ~2 s after
start, and handed an empty list to NewFilerClientAccessor, which panics. A
fresh filer registers only after its empty-membership confirm window
(10 s), so a gateway started with a fresh cluster crashed (Kafka Quick
Test, kafka-gateway exit 2). Retry while the list is empty for up to 60 s,
then return an error.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* mq/kafka: wait out a transient-empty broker listing too

A fresh SMQ broker registers with the masters on the same announce
delay as a filer, so the one-shot broker lookup could fail startup the
same way. Generalize waitForFilers into waitForClusterNodes and bound
both discoveries by the same window; the empty-broker check becomes
unreachable and drops away.

* mq/kafka: wait out the broker health check like node discovery

A broker announces to the masters before it has acquired its balancer
lock, so the gateway's three-attempt probe could still fail fast on a
fresh cluster. Retry FindBrokerLeader for the same window the caller
waits out empty listings.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: user.email <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/mq/kafka/integration/broker_client.go

@@ -22,9 +22,8 @@ import (
 )
 
 const (
-	brokerHealthCheckTimeout  = 5 * time.Second
-	brokerHealthCheckAttempts = 3
-	brokerHealthCheckBackoff  = 250 * time.Millisecond
+	brokerHealthCheckTimeout = 5 * time.Second
+	brokerHealthCheckBackoff = 250 * time.Millisecond
 )
 
 // NewBrokerClientWithFilerAccessor creates a client with a shared filer accessor
@@ -102,14 +101,17 @@ func (bc *BrokerClient) Close() error {
 	return bc.conn.Close()
 }
 
-// HealthCheck verifies the broker connection is working
+// HealthCheck verifies the broker connection is working. A freshly announced
+// broker may still be acquiring its balancer lock, so keep retrying for the
+// same window cluster discovery waits out.
 func (bc *BrokerClient) HealthCheck() error {
 	if bc.client == nil {
 		return fmt.Errorf("broker client not connected")
 	}
 
+	deadline := time.Now().Add(clusterNodeDiscoveryTimeout)
 	var lastErr error
-	for attempt := 1; attempt <= brokerHealthCheckAttempts; attempt++ {
+	for attempt := 1; ; attempt++ {
 		ctx, cancel := context.WithTimeout(bc.ctx, brokerHealthCheckTimeout)
 		_, err := bc.client.FindBrokerLeader(ctx, &mq_pb.FindBrokerLeaderRequest{})
 		cancel()
@@ -117,17 +119,18 @@ func (bc *BrokerClient) HealthCheck() error {
 			return nil
 		}
 		lastErr = err
-		if attempt == brokerHealthCheckAttempts {
+		if !time.Now().Before(deadline) {
 			break
 		}
+		backoff := min(time.Duration(attempt)*brokerHealthCheckBackoff, time.Second)
 		select {
 		case <-bc.ctx.Done():
 			return bc.ctx.Err()
-		case <-time.After(time.Duration(attempt) * brokerHealthCheckBackoff):
+		case <-time.After(backoff):
 		}
 	}
 
-	return fmt.Errorf("broker health check failed after %d attempts: %v", brokerHealthCheckAttempts, lastErr)
+	return fmt.Errorf("broker health check failed after %v: %v", clusterNodeDiscoveryTimeout, lastErr)
 }
 
 // GetPartitionRangeInfo gets comprehensive range information from SeaweedMQ broker's native range manager

weed/mq/kafka/integration/broker_client_test.go

@@ -0,0 +1,46 @@
+package integration
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+)
+
+type stubMessagingClient struct {
+	mq_pb.SeaweedMessagingClient
+	calls   int
+	failFor int
+	err     error
+}
+
+func (s *stubMessagingClient) FindBrokerLeader(ctx context.Context, in *mq_pb.FindBrokerLeaderRequest, opts ...grpc.CallOption) (*mq_pb.FindBrokerLeaderResponse, error) {
+	s.calls++
+	if s.calls <= s.failFor {
+		return nil, s.err
+	}
+	return &mq_pb.FindBrokerLeaderResponse{}, nil
+}
+
+func TestHealthCheckRetriesPastTransientBrokerErrors(t *testing.T) {
+	stub := &stubMessagingClient{failFor: 2, err: errors.New("lock broker_balancer not found")}
+	bc := &BrokerClient{ctx: context.Background(), client: stub}
+	if err := bc.HealthCheck(); err != nil || stub.calls != 3 {
+		t.Fatalf("err %v after %d calls, want success on the 3rd", err, stub.calls)
+	}
+}
+
+func TestHealthCheckGivesUpAtDeadline(t *testing.T) {
+	defer func(orig time.Duration) { clusterNodeDiscoveryTimeout = orig }(clusterNodeDiscoveryTimeout)
+	clusterNodeDiscoveryTimeout = 300 * time.Millisecond
+
+	stub := &stubMessagingClient{failFor: 1 << 30, err: errors.New("lock broker_balancer not found")}
+	bc := &BrokerClient{ctx: context.Background(), client: stub}
+	if err := bc.HealthCheck(); err == nil || stub.calls < 2 {
+		t.Fatalf("err %v after %d calls, want repeated retries then an error", err, stub.calls)
+	}
+}

weed/mq/kafka/integration/seaweedmq_handler_utils.go

@@ -52,21 +52,23 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str
 	time.Sleep(2 * time.Second)
 	glog.V(1).Infof("Initial connection delay completed")
 
-	// Discover brokers from masters using master client
+	// Discover brokers and filers from masters using master client. A fresh
+	// node registers only after its announce delay (a filer also waits out its
+	// empty-membership confirm window), so an empty list right after startup
+	// is normal: wait rather than fail fast.
 	glog.V(1).Infof("About to call discoverBrokersWithMasterClient...")
-	brokerAddresses, err := discoverBrokersWithMasterClient(ctx, masterClient, filerGroup)
+	brokerAddresses, err := waitForClusterNodes(ctx, "brokers", clusterNodeDiscoveryTimeout, func() ([]string, error) {
+		return discoverBrokersWithMasterClient(ctx, masterClient, filerGroup)
+	})
 	if err != nil {
 		glog.Errorf("Broker discovery failed: %v", err)
 		return nil, fmt.Errorf("failed to discover brokers: %v", err)
 	}
 	glog.V(1).Infof("Broker discovery returned: %v", brokerAddresses)
 
-	if len(brokerAddresses) == 0 {
-		return nil, fmt.Errorf("no brokers discovered from masters")
-	}
-
-	// Discover filers from masters using master client
-	filerAddresses, err := discoverFilersWithMasterClient(ctx, masterClient, filerGroup)
+	filerAddresses, err := waitForClusterNodes(ctx, "filers", clusterNodeDiscoveryTimeout, func() ([]pb.ServerAddress, error) {
+		return discoverFilersWithMasterClient(ctx, masterClient, filerGroup)
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to discover filers: %v", err)
 	}
@@ -143,7 +145,31 @@ func discoverBrokersWithMasterClient(ctx context.Context, masterClient *wdclient
 	return brokers, err
 }
 
-// discoverFilersWithMasterClient queries masters for available filers using reusable master client
+// clusterNodeDiscoveryTimeout bounds how long the gateway waits for the
+// masters to list a node; well above a fresh node's announce delay.
+var clusterNodeDiscoveryTimeout = 60 * time.Second
+
+// waitForClusterNodes calls discover until it returns at least one node, an
+// error, or timeout passes, retrying once a second while the list is empty.
+func waitForClusterNodes[T any](ctx context.Context, nodeType string, timeout time.Duration, discover func() ([]T, error)) ([]T, error) {
+	deadline := time.Now().Add(timeout)
+	for {
+		nodes, err := discover()
+		if err != nil || len(nodes) > 0 {
+			return nodes, err
+		}
+		if !time.Now().Before(deadline) {
+			return nil, fmt.Errorf("no %s registered with the masters after %v", nodeType, timeout)
+		}
+		glog.V(1).Infof("no %s registered with the masters yet; retrying", nodeType)
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(time.Second):
+		}
+	}
+}
+
 func discoverFilersWithMasterClient(ctx context.Context, masterClient *wdclient.MasterClient, filerGroup string) ([]pb.ServerAddress, error) {
 	var filers []pb.ServerAddress
 

weed/mq/kafka/integration/seaweedmq_handler_utils_wait_test.go

@@ -0,0 +1,38 @@
+package integration
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// A fresh filer is listed by the masters only after its announce delay, so an
+// empty list must be waited out, not handed to the filer client accessor
+// (which panics on it).
+func TestWaitForFilersWaitsOutEmptyListing(t *testing.T) {
+	calls := 0
+	filers, err := waitForClusterNodes(context.Background(), "filers", 10*time.Second, func() ([]pb.ServerAddress, error) {
+		calls++
+		if calls < 3 {
+			return nil, nil
+		}
+		return []pb.ServerAddress{"filer:8888"}, nil
+	})
+	if err != nil || len(filers) != 1 || calls != 3 {
+		t.Fatalf("filers %v err %v after %d calls, want 1 filer after 3", filers, err, calls)
+	}
+}
+
+func TestWaitForFilersGivesUpWithAnError(t *testing.T) {
+	filers, err := waitForClusterNodes(context.Background(), "filers", 1500*time.Millisecond, func() ([]pb.ServerAddress, error) { return nil, nil })
+	if err == nil || len(filers) != 0 {
+		t.Fatalf("still empty: filers %v err %v, want an error", filers, err)
+	}
+	boom := errors.New("master down")
+	if _, err := waitForClusterNodes(context.Background(), "filers", time.Minute, func() ([]pb.ServerAddress, error) { return nil, boom }); !errors.Is(err, boom) {
+		t.Fatalf("discovery error: got %v, want it returned at once", err)
+	}
+}