apache/beam · error

control.Recv failed

Error message

control.Recv failed

What it means

The harness receives instruction requests from the control stream in a loop via stub.Recv(). On any error other than a clean io.EOF (which means the server ended the stream normally, e.g. graceful shutdown), the loop exits and wraps the error as "control.Recv failed". It indicates the control stream broke mid-run, typically due to a network interruption or a server-side failure while executing bundles.

Source

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

		}
	}

	// gRPC requires all readers of a stream be the same goroutine, so this goroutine
	// is responsible for managing the network data. All it does is pull data from
	// the stream, and hand off the message to a goroutine to actually be handled,
	// so as to avoid blocking the underlying network channel.
	var shutdown int32
	for {
		req, err := stub.Recv()
		if err != nil {
			// An error means we can't send or receive anymore. Shut down.
			atomic.AddInt32(&shutdown, 1)
			close(respc)
			wg.Wait()
			if err == io.EOF {
				return nil
			}
			return errors.Wrapf(err, "control.Recv failed")
		}

		// Launch a goroutine to handle the control message.
		fn := func(ctx context.Context, req *fnpb.InstructionRequest) {
			// TODO(lostluck): 2023/03/29 fix debug level logging to be flagged.
			// log.Debugf(ctx, "RECV: %v", proto.MarshalTextString(req))
			ctx = hooks.RunRequestHooks(ctx, req)
			resp := ctrl.handleInstruction(ctx, req)

			hooks.RunResponseHooks(ctx, req, resp)

			if resp != nil && atomic.LoadInt32(&shutdown) == 0 {
				respc <- resp
			}
		}

		if req.GetProcessBundle() != nil {
			// Add this to the inactive queue before allowing other requests

View on GitHub (pinned to 12126d8942)

Solutions

1. Inspect the wrapped gRPC status to distinguish Canceled (normal shutdown), Unavailable/DeadlineExceeded (network), or Internal (server crash).
2. Enable gRPC keepalives and raise LB idle timeouts so long-quiet streams are not dropped.
3. Ensure the runner/control service has enough memory; check for OOM kills around the failure time.
4. Re-run the pipeline; if it fails at the same bundle, investigate that bundle/transform for server-side crashes.
5. Upgrade runner+SDK to versions with resilient reconnect handling for the control stream.

Example fix

// server-side: keep long-lived streams alive
srv := grpc.NewServer(
    grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
        MinTime: 10 * time.Second, PermitWithoutStream: true,
    }),
)
// harness side: ensure ctx is not cancelled prematurely and idle timeouts exceed max bundle duration

if err := runHarness(ctx, opts); err != nil {
    if strings.Contains(err.Error(), "control.Recv failed") {
        code := status.Code(errors.Unwrap(err))
        if code == codes.Canceled {
            return nil // graceful shutdown path
        }
        // otherwise: connection dropped mid-run — alert/inspect network and server health
    }
    return err
}

When it happens

Trigger: stub.Recv() returns a non-EOF error during the receive loop: stream reset, connection dropped, server crashed, context cancelled abnormally, or a gRPC deadline exceeded while waiting for the next instruction.

Common situations: Long-running pipelines where the control connection is dropped by load-balancer idle timeouts; runner master restarts mid-job; network partition between worker and runner; OOM-killed control server during heavy bundles.

Related errors

• failed to connect
• failed to connect to control service
• Logging stream terminated unexpectedly before it was closed
• Failed to call Rate Limit Service
• Failed to get response from Rate Limit Service

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