apache/beam · error

failed to send chunks for %v; close error: %v

Error message

failed to send chunks for %v; close error: %v

What it means

Stage() in the Beam Go artifact package uploads a local file to the LegacyArtifactStagingService over a gRPC client stream. After the metadata header is sent, stageChunks streams 1MB data chunks; if any chunk Send fails (or reads fail), Stage tries stream.CloseAndRecv() to drain the stream and wraps the original error with the close error appended. This message means the artifact upload aborted mid-stream and reports both why the send failed and whether closing the stream also failed.

Source

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

	stream, err := client.PutArtifact(ctx)
	if err != nil {
		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

View on GitHub (pinned to 12126d8942)

Solutions

1. Inspect the wrapped root cause (err) first — it names the gRPC status (Unavailable, Canceled, DeadlineExceeded) that actually killed the upload; fix that underlying condition.
2. Verify the artifact staging service endpoint is reachable and the staging session token is valid; an invalid token causes the server to cancel the stream after the header is sent.
3. Increase the gRPC context timeout / check network stability for large files; MultiStage already retries 3 times with jittered backoff, so persistent failures indicate an environment problem, not transient load.
4. If errClose is non-nil too, check server logs: it means the server also failed during close, typically because the stream was already terminated server-side.

Example fix

// before
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)
}
// after (caller-side: surface and classify the root cause before retrying)
stagedHash, err := stageChunks(stream, fd)
if err != nil {
	_, errClose := stream.CloseAndRecv()
	if status.Code(errors.Unwrap(err)) == codes.Unavailable || status.Code(errors.Unwrap(err)) == codes.DeadlineExceeded {
		return nil, retryable.New(errors.Wrapf(err, "failed to send chunks for %v; close error: %v", filename, errClose))
	}
	return nil, errors.Wrapf(err, "failed to send chunks for %v; close error: %v", filename, errClose)
}

conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil { return err }
client := jobpb.NewLegacyArtifactStagingServiceClient(conn)
// pre-check service reachability before staging
c, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
if _, err := client.PutArtifact(c); err != nil {
	return fmt.Errorf("artifact staging service unreachable: %w", err)
}

if _, err := artifact.MultiStage(ctx, client, 10, files, token); err != nil {
	var gerr errorx.GuardedError // persistent failure after 3 attempts
	log.Printf("staging failed permanently: %v", err)
	// inspect wrapped grpc status before deciding to retry at a higher level
}

When it happens

Trigger: stream.Send(chunk) inside stageChunks returns a non-EOF error (server cancelled the PutArtifact RPC, connection dropped, deadline exceeded, or the local file read failed via the r.Read error path only if not EOF). The CloseAndRecv errClose is interpolated but is secondary.

Common situations: Runner/artifact staging service went away mid-upload (job cancelled or crashed); network interruption or gRPC context deadline exceeded while transferring a large artifact; server rejects the session token and cancels the stream after the header; transient flakiness that MultiStage retries up to 3 times before giving up with 'failed to stage %v in 3 attempts'.

Related errors

• chunk send failed
• failed to close stream for %v; response: %v
• failed to receive header
• failed to retrieve chunk for %v
• failed to connect to state service %v

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