diff --git a/weed/mq/kafka/integration/broker_client.go b/weed/mq/kafka/integration/broker_client.go index 3c3482c86e..428cfdffef 100644 --- a/weed/mq/kafka/integration/broker_client.go +++ b/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 diff --git a/weed/mq/kafka/integration/broker_client_test.go b/weed/mq/kafka/integration/broker_client_test.go new file mode 100644 index 0000000000..efcb8c9153 --- /dev/null +++ b/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) + } +} diff --git a/weed/mq/kafka/integration/seaweedmq_handler_utils.go b/weed/mq/kafka/integration/seaweedmq_handler_utils.go index 33c3590ff4..e17cf4eb69 100644 --- a/weed/mq/kafka/integration/seaweedmq_handler_utils.go +++ b/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 diff --git a/weed/mq/kafka/integration/seaweedmq_handler_utils_wait_test.go b/weed/mq/kafka/integration/seaweedmq_handler_utils_wait_test.go new file mode 100644 index 0000000000..ac3c03115c --- /dev/null +++ b/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) + } +}