apache/beam · error

failed to connect: %v

Error message

failed to connect: %v

What it means

newWorkerStatusHandler dials the runner's worker status gRPC endpoint (60s timeout) to stream status/liveness info. If the dial fails, it wraps the error with the endpoint. Without this connection the harness cannot serve worker status, though the pipeline may still process data.

Source

Thrown at sdks/go/pkg/beam/core/runtime/harness/worker_status.go:50

	"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
	fnpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/fnexecution_v1"
	"google.golang.org/grpc"
)

// workerStatusHandler stores the communication information of WorkerStatus API.
type workerStatusHandler struct {
	conn             *grpc.ClientConn
	shouldShutdown   int32
	wg               sync.WaitGroup
	cache            *statecache.SideInputCache
	metStoreToString func(*strings.Builder)
}

func newWorkerStatusHandler(ctx context.Context, endpoint string, cache *statecache.SideInputCache, metStoreToString func(*strings.Builder)) (*workerStatusHandler, error) {
	sconn, err := dial(ctx, endpoint, "status", 60*time.Second)
	if err != nil {
		return nil, errors.Wrapf(err, "failed to connect: %v\n", endpoint)
	}
	return &workerStatusHandler{conn: sconn, shouldShutdown: 0, cache: cache, metStoreToString: metStoreToString}, nil
}

func (w *workerStatusHandler) isAlive() bool {
	return atomic.LoadInt32(&w.shouldShutdown) == 0
}

func (w *workerStatusHandler) shutdown() {
	atomic.StoreInt32(&w.shouldShutdown, 1)
}

// start starts the reader to accept WorkerStatusRequest and send WorkerStatusResponse with WorkerStatus API.
func (w *workerStatusHandler) start(ctx context.Context) error {
	statusClient := fnpb.NewBeamFnWorkerStatusClient(w.conn)
	stub, err := statusClient.WorkerStatus(ctx)
	if err != nil {
		log.Errorf(ctx, "status client not established: %v", err)

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the status endpoint URL the runner provisioned is reachable from the worker (curl/nc the host:port).
2. Check DNS resolution and network policies/firewalls on the worker subnet.
3. Check TLS settings match the runner's gRPC server configuration.
4. If the runner starts the status service late, retry or increase the dial budget; if the runner lacks status support, ensure it provides the endpoint or handle startup failure gracefully.

if endpoint == "" { return errors.New("worker status endpoint not provisioned") }

h, err := newWorkerStatusHandler(ctx, endpoint, cache, metToString)
if err != nil {
    log.Warnf("status handler unavailable: %v (continuing without status)", err)
    h = nil // degrade gracefully when status is optional
}

When it happens

Trigger: MainWithOptions starting a harness whose status endpoint is unreachable: wrong/empty endpoint URL from the runner provisioning, DNS failure, status service not yet up, TLS mismatch, or firewall blocking the port.

Common situations: Misconfigured statusApiServiceDescriptor in the runner, slow runner startup exceeding the 60s dial window in constrained environments, Kubernetes/network policies blocking harness→runner traffic, custom runners that omit status service support.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• failed to send chunks for %v; close error: %v
• chunk send failed
• failed to connect to state service %v
• StateChannel[%v].Send(%v): channel closed due to: %v
• StageFile chunk send failed

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/66168c5b234f91c7. Report an issue: GitHub.