apache/beam · error

GroupIntoBatchesWithShardedKey: %w

Error message

GroupIntoBatchesWithShardedKey: %w

What it means

GroupIntoBatchesWithShardedKey panics when params.validate() fails, wrapping the underlying Params error with %w. Same validation rules as GroupIntoBatches: no negative sizes/durations and at least one of BatchSize/BatchSizeBytes > 0. The wrapped cause names the offending field.

Source

Thrown at sdks/go/pkg/beam/transforms/batch/batch.go:660

	return beam.ParDo(s, fn, col)
}

// GroupIntoBatchesWithShardedKey wraps each user key with a
// ShardedKey{Key: K, ShardID: [24]byte} and then applies
// GroupIntoBatches. Output is PCollection<KV<ShardedKey[K], []V>>.
//
// The key type K must have been registered via
// RegisterShardedKeyType[K] at init time. Common types (string,
// []byte, int, int64) are registered automatically.
//
// Sharding spreads the processing of a single hot logical key across
// multiple workers: each shard is independent state, so distributed
// runners can parallelize without the user's key type changing.
func GroupIntoBatchesWithShardedKey[K any](s beam.Scope, params Params, col beam.PCollection) beam.PCollection {
	s = s.Scope("batch.GroupIntoBatchesWithShardedKey")

	if err := params.validate(); err != nil {
		panic(fmt.Errorf("GroupIntoBatchesWithShardedKey: %w", err))
	}
	if !typex.IsKV(col.Type()) {
		panic(fmt.Errorf(
			"GroupIntoBatchesWithShardedKey: input PCollection must be KV-typed; got %v",
			col.Type()))
	}
	keyFT := col.Type().Components()[0]
	var zero K
	if keyFT.Type() != reflect.TypeOf(zero) {
		panic(fmt.Errorf(
			"GroupIntoBatchesWithShardedKey: type parameter K (%v) does not match input key type (%v)",
			reflect.TypeOf(zero), keyFT.Type()))
	}

	wrapped := beam.ParDo(s, &wrapShardedKeyFn[K]{}, col)
	return GroupIntoBatches(s, params, wrapped)
}

View on GitHub (pinned to 12126d8942)

Solutions

1. Inspect the wrapped validate error in the panic message
2. Set BatchSize > 0 or BatchSizeBytes > 0
3. Ensure no Params fields are negative
4. Validate Params before building the pipeline

Example fix

// before
params := batch.Params{} // invalid: no limits
// after
params := batch.Params{BatchSizeBytes: 1 << 20, BatchSize: 1000}

if err := params.validate(); err != nil { return fmt.Errorf("sharded batching: %w", err) }

When it happens

Trigger: Calling GroupIntoBatchesWithShardedKey[K](s, params, col) with invalid Params (zero limits, negative BatchSizeBytes, negative MaxBufferingDuration).

Common situations: Reusing default-constructed Params{} with no limits; config flags that set both limits to 0 to 'disable' batching.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• GroupIntoBatches: %w
• trigger.AfterCount(%v) must be a positive integer
• can't apply processing delay of less than a millisecond. Got
• can't apply an alignment period of less than a millisecond.
• broken stream

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