apache/beam · error

GroupIntoBatches: %w

Error message

GroupIntoBatches: %w

What it means

GroupIntoBatches panics when params.validate() reports any invalid Params field (negative sizes, no size/byte limit, negative buffering duration). The underlying validate error is wrapped with %w and raised as a panic at pipeline build time. The panic message embeds the specific field problem.

Source

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

	RegisterShardedKeyType[string]()
	RegisterShardedKeyType[int]()
	RegisterShardedKeyType[int64]()
}

// GroupIntoBatches groups the values of the input PCollection<KV<K, V>>
// into batches of up to params.BatchSize elements (or
// params.BatchSizeBytes bytes) per key and emits them as
// PCollection<KV<K, []V>>.
//
// The input must be KV-typed. The key coder must be deterministic;
// non-deterministic key coders would corrupt state keying. Panics at
// pipeline build time on invalid params, non-KV input, zero limits, or
// a non-deterministic key coder.
func GroupIntoBatches(s beam.Scope, params Params, col beam.PCollection) beam.PCollection {
	s = s.Scope("batch.GroupIntoBatches")

	if err := params.validate(); err != nil {
		panic(fmt.Errorf("GroupIntoBatches: %w", err))
	}
	if !typex.IsKV(col.Type()) {
		panic(fmt.Errorf(
			"GroupIntoBatches: input PCollection must be KV-typed; got %v", col.Type()))
	}

	keyFT := col.Type().Components()[0]
	valFT := col.Type().Components()[1]

	if !beam.NewCoder(keyFT).IsDeterministic() {
		panic(fmt.Errorf(
			"GroupIntoBatches: key coder for type %v is not deterministic. "+
				"Register a deterministic custom coder with "+
				"coder.RegisterDeterministicCoder, or use a deterministic key "+
				"type (string, []byte, bool, integer, float)", keyFT.Type()))
	}

	sizerKind := sizerNone

View on GitHub (pinned to 12126d8942)

Solutions

1. Read the wrapped cause in the panic text to see which field failed
2. Set BatchSize > 0 or BatchSizeBytes > 0 (at least one limit required)
3. Ensure no limit/buffer fields are negative
4. Validate Params before entering pipeline construction

Example fix

// before
params := batch.Params{} // no limits set
// after
params := batch.Params{BatchSize: 500}

if params.BatchSize <= 0 && params.BatchSizeBytes <= 0 {
    return errors.New("set BatchSize or BatchSizeBytes > 0")
}
if err := params.validate(); err != nil { return err }

When it happens

Trigger: Calling GroupIntoBatches(s, params, col) with a Params whose BatchSizeBytes < 0, MaxBufferingDuration < 0, or with both BatchSize == 0 and BatchSizeBytes == 0.

Common situations: Building Params from user-supplied flags or YAML where limits are unset (zero) or negative; upgrading Beam and relying on previously-permitted defaults.

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

• GroupIntoBatchesWithShardedKey: %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/7fd1f67ada2df5fe. Report an issue: GitHub.