apache/beam · error

failed to close stream for %v; response: %v

Error message

failed to close stream for %v; response: %v

What it means

Stage() closes the PutArtifact gRPC stream with CloseAndRecv() to get the server's final response. If the close/receive returns an error other than io.EOF, Stage wraps it as 'failed to close stream for %v; response: %v'. This means the server did not cleanly finish the artifact upload — the final RPC handshake failed even though all chunks may have been sent.

Source

Thrown at sdks/go/pkg/beam/artifact/stage.go:166

		return nil, err
	}

	header := &jobpb.PutArtifactRequest{
		Content: &jobpb.PutArtifactRequest_Metadata{
			Metadata: pmd,
		},
	}
	if err := stream.Send(header); err != nil {
		stream.CloseAndRecv() // ignore error
		return nil, errors.Wrapf(err, "failed to send header for %v", filename)
	}
	stagedHash, err := stageChunks(stream, fd)
	if err != nil {
		_, errClose := stream.CloseAndRecv()
		return nil, errors.Wrapf(err, "failed to send chunks for %v; close error: %v", filename, errClose)
	}
	if resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {
		return nil, errors.Wrapf(err, "failed to close stream for %v; response: %v", filename, resp)
	}
	if hash != stagedHash {
		return nil, errors.Errorf("unexpected SHA256 for sent chunks for %v: %v, want %v", filename, stagedHash, hash)
	}
	return md, nil
}

func stageChunks(stream jobpb.LegacyArtifactStagingService_PutArtifactClient, r io.Reader) (string, error) {
	sha256W := sha256.New()
	data := make([]byte, 1<<20)
	for {
		n, err := r.Read(data)
		if n > 0 {
			if _, err := sha256W.Write(data[:n]); err != nil {
				panic(err) // cannot fail
			}

			chunk := &jobpb.PutArtifactRequest{

View on GitHub (pinned to 12126d8942)

Solutions

1. Check the wrapped gRPC status code: io.EOF is treated as success here, any other status is a real server-side rejection or transport failure.
2. Verify the staging session token is valid and not expired — servers commonly reject at CloseAndRecv time when finalizing.
3. Increase the context deadline passed to Stage/MultiStage for large artifacts, or reduce artifact size.
4. Confirm the artifact staging service is healthy and stable for the whole upload duration; retry via MultiStage (built-in 3 attempts) only for transient statuses like Unavailable.

Example fix

// before
if resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {
	return nil, errors.Wrapf(err, "failed to close stream for %v; response: %v", filename, resp)
}
// after (caller-side: check gRPC status to distinguish transient vs fatal)
if resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {
	switch status.Code(errors.Unwrap(err)) {
	case codes.Unavailable, codes.DeadlineExceeded:
		return nil, retryable.New(errors.Wrapf(err, "failed to close stream for %v; response: %v", filename, resp))
	default:
		return nil, errors.Wrapf(err, "failed to close stream for %v; response: %v", filename, resp)
	}
}

if _, err := artifact.Stage(ctx, client, key, filename, token); err != nil {
	if code := status.Code(errors.Unwrap(err)); code == codes.Unavailable || code == codes.DeadlineExceeded {
		// transient: retry with backoff
	} else {
		return fmt.Errorf("staging close rejected (code=%v): %w", code, err)
	}
}

When it happens

Trigger: stream.CloseAndRecv() after successful chunk sends returns a non-EOF error: server returned a gRPC status error (e.g. InvalidArgument, permission/token check failure at commit time, Unavailable), the context was cancelled, or the connection was reset before the server's response arrived. resp (usually nil on error) is interpolated into the message.

Common situations: Staging session token rejected when the server finalizes the artifact; artifact service crashed or was redeployed between chunk upload and close; per-request context deadline exceeded for very large artifacts; job pipeline cancelled while staging was in flight.

Related errors

• failed to send chunks for %v; close error: %v
• chunk send failed
• failed to get job stream
• Logging stream terminated unexpectedly with success before i
• error creating local job server: %v

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