From d64228418b7885f6ffa94d60f3265d33d64e7c83 Mon Sep 17 00:00:00 2001 From: Jiale Lin <63439129+ljluestc@users.noreply.github.com> Date: Sat, 21 Mar 2026 10:13:10 -0700 Subject: [PATCH] fix(pulsar): increase wait strategy deadlines to fix random test timeouts The Pulsar module used WithWaitStrategy which has a default 60s deadline. Pulsar is a heavy Java application that can take longer than 60s to start, especially in CI environments, causing random context deadline exceeded failures. Changes: - Use WithWaitStrategyAndDeadline(2min) for default Run() - Use WithWaitStrategyAndDeadline(3min) for WithFunctionsWorker() - Use WithWaitStrategyAndDeadline(2min) for WithTransactions() - Add explicit WithStartupTimeout(2min) to HTTP wait strategies Fixes #2886 Co-Authored-By: Oz --- modules/pulsar/pulsar.go | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/modules/pulsar/pulsar.go b/modules/pulsar/pulsar.go index 4cd1ca121d..a762114dd5 100644 --- a/modules/pulsar/pulsar.go +++ b/modules/pulsar/pulsar.go @@ -6,6 +6,7 @@ import ( "io" "net" "strings" + "time" "github.com/docker/go-connections/nat" @@ -22,7 +23,7 @@ const ( ) var defaultWaitStrategies = []wait.Strategy{ - wait.ForHTTP("/admin/v2/clusters").WithPort(defaultPulsarAdminPort).WithResponseMatcher(func(r io.Reader) bool { + wait.ForHTTP("/admin/v2/clusters").WithPort(defaultPulsarAdminPort).WithStartupTimeout(2 * time.Minute).WithResponseMatcher(func(r io.Reader) bool { respBytes, _ := io.ReadAll(r) resp := string(respBytes) return resp == `["standalone"]` @@ -80,7 +81,7 @@ func WithFunctionsWorker() testcontainers.CustomizeRequestOption { ss = append(ss, wait.ForLog("Function worker service started")) ss = append(ss, defaultWaitStrategies...) - return testcontainers.WithWaitStrategy(ss...)(req) + return testcontainers.WithWaitStrategyAndDeadline(3*time.Minute, ss...)(req) } } @@ -113,12 +114,12 @@ func WithTransactions() testcontainers.CustomizeRequestOption { // clone defaultWaitStrategies ss := make([]wait.Strategy, 0, 1+len(defaultWaitStrategies)) - ss = append(ss, wait.ForHTTP(transactionTopicEndpoint).WithPort(defaultPulsarAdminPort).WithStatusCodeMatcher(func(statusCode int) bool { + ss = append(ss, wait.ForHTTP(transactionTopicEndpoint).WithPort(defaultPulsarAdminPort).WithStartupTimeout(2*time.Minute).WithStatusCodeMatcher(func(statusCode int) bool { return statusCode == 200 })) ss = append(ss, defaultWaitStrategies...) - return testcontainers.WithWaitStrategy(ss...)(req) + return testcontainers.WithWaitStrategyAndDeadline(2*time.Minute, ss...)(req) } } @@ -141,7 +142,7 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom moduleOpts := make([]testcontainers.ContainerCustomizer, 0, 3+len(opts)) moduleOpts = append(moduleOpts, testcontainers.WithExposedPorts(defaultPulsarPort, defaultPulsarAdminPort), - testcontainers.WithWaitStrategy(defaultWaitStrategies...), + testcontainers.WithWaitStrategyAndDeadline(2*time.Minute, defaultWaitStrategies...), testcontainers.WithCmd("/bin/bash", "-c", strings.Join([]string{defaultPulsarCmd, defaultPulsarCmdWithoutFunctionsWorker}, " ")), )