What causes "failed to connect" in apache/beam?

Verify the control endpoint host:port is correct and reachable from the worker (nc/curl from inside the worker container).

apache/beam · critical

failed to connect

Error message

failed to connect

What it means

In Apache Beam Go SDK's FnAPI harness (MainWithOptions), the worker dials the control gRPC server at controlEndpoint with a 60s timeout. If dialing fails for any reason (unreachable host, TLS rejection, timeout), the underlying error is wrapped as "failed to connect" and worker startup aborts. This is the first network step of the Go SDK harness, so it usually reflects container networking or endpoint configuration problems.

Source

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

		}
		if err := profiler.Start(cfg); err != nil {
			log.Errorf(ctx, "failed to start cloud profiler, got %v", err)
		}
	}

	if tempLocation := beam.PipelineOptions.Get("temp_location"); tempLocation != "" && samplingFrequencySeconds > 0 {
		go diagnostics.SampleForHeapProfile(ctx, samplingFrequencySeconds, maxTimeBetweenDumpsSeconds)
	}

	elmTimeout, err := parseTimeoutDurationFlag(ctx, beam.PipelineOptions.Get("element_processing_timeout"))
	if err != nil {
		log.Debugf(ctx, "Failed to parse element_processing_timeout: %v, there will be no timeout for processing an element in a PTransform operation", err)
	}

	// Connect to FnAPI control server. Receive and execute work.
	conn, err := dial(ctx, controlEndpoint, "control", 60*time.Second)
	if err != nil {
		return errors.Wrap(err, "failed to connect")
	}
	defer conn.Close()

	client := fnpb.NewBeamFnControlClient(conn)

	lookupDesc := func(id bundleDescriptorID) (*fnpb.ProcessBundleDescriptor, error) {
		return client.GetProcessBundleDescriptor(ctx, &fnpb.GetProcessBundleDescriptorRequest{ProcessBundleDescriptorId: string(id)})
	}

	stub, err := client.Control(ctx)
	if err != nil {
		return errors.Wrapf(err, "failed to connect to control service")
	}

	log.Debugf(ctx, "Successfully connected to control @ %v", controlEndpoint)

	// Each ProcessBundle is a sub-graph of the original one.

View on GitHub (pinned to 12126d8942)

Solutions

Verify the control endpoint host:port is correct and reachable from the worker (nc/curl from inside the worker container).
Ensure the FnAPI control/runner service is started before the SDK harness launches; tolerate startup races with retries.
Check network policy/firewall/VPC rules allow worker-to-control traffic on the gRPC port.
Re-run with debug logging to see the wrapped underlying dial error and fix its root cause (DNS, TLS, auth).
Pin/upgrade the Beam SDK and runner versions so the FnAPI endpoints and proto versions match.

Example fix

// before: relying on a stale/hardcoded endpoint
conn, err := dial(ctx, controlEndpoint, "control", 60*time.Second)
if err != nil {
    return errors.Wrap(err, "failed to connect")
}
// after: validate the endpoint before dialing
if controlEndpoint == "" {
    return errors.New("control endpoint not provided")
}
if _, _, err := net.SplitHostPort(string(controlEndpoint)); err != nil {
    return errors.Wrapf(err, "invalid control endpoint %q", controlEndpoint)
}
conn, err := dial(ctx, controlEndpoint, "control", 60*time.Second)
if err != nil {
    return errors.Wrap(err, "failed to connect")
}

Defensive patterns

Strategy: retry

Validate before calling

func validateControlEndpoint(ep string) error {
    host, port, err := net.SplitHostPort(ep)
    if err != nil || host == "" || port == "" {
        return fmt.Errorf("invalid control endpoint %q", ep)
    }
    conn, err := net.DialTimeout("tcp", ep, 5*time.Second)
    if err != nil {
        return fmt.Errorf("control endpoint %q unreachable: %v", ep, err)
    }
    conn.Close()
    return nil
}

Try / catch

err := runHarness(ctx, opts)
var ne net.Error
switch {
case errors.As(err, &ne) && ne.Timeout():
    // endpoint unreachable within 60s: check networking/service startup, retry with backoff
case status.Code(errors.Unwrap(err)) == codes.Unavailable:
    // transient; retry worker startup
default:
    return err
}

Prevention

Verify worker-to-control connectivity (host, port, firewall, NetworkPolicy) before launching jobs with custom containers.
Keep runner and Beam SDK versions matched so endpoints and FnAPI protos are compatible.
Make sure the control service starts before SDK workers, or tolerate initial retries.
Test custom containers locally against a runner (e.g. DirectRunner/Flink) before production jobs.

When it happens

Trigger: dial(ctx, controlEndpoint, "control", 60*time.Second) fails: control service address unreachable, DNS/hostname unresolvable, TLS/credentials rejected, connection refused, or the 60-second deadline expires before the gRPC channel is established.

Common situations: Runner-provided --control_endpoint or worker options point to a wrong host/port; the FnAPI control service container isn't up yet (startup race in Docker/Flink/Spark runners); firewall or Kubernetes NetworkPolicy blocks the port; grpc:// vs TLS implications with custom containers; endpoints behind VPC without private access.

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

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