{"record":{"id":"a9b50102b3bdb823","repo":"apache/beam","slug":"groupintobatcheswithshardedkey-w","errorCode":null,"errorMessage":"GroupIntoBatchesWithShardedKey: %w","messagePattern":"GroupIntoBatchesWithShardedKey: %w","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/transforms/batch/batch.go","lineNumber":660,"sourceCode":"\treturn beam.ParDo(s, fn, col)\n}\n\n// GroupIntoBatchesWithShardedKey wraps each user key with a\n// ShardedKey{Key: K, ShardID: [24]byte} and then applies\n// GroupIntoBatches. Output is PCollection<KV<ShardedKey[K], []V>>.\n//\n// The key type K must have been registered via\n// RegisterShardedKeyType[K] at init time. Common types (string,\n// []byte, int, int64) are registered automatically.\n//\n// Sharding spreads the processing of a single hot logical key across\n// multiple workers: each shard is independent state, so distributed\n// runners can parallelize without the user's key type changing.\nfunc GroupIntoBatchesWithShardedKey[K any](s beam.Scope, params Params, col beam.PCollection) beam.PCollection {\n\ts = s.Scope(\"batch.GroupIntoBatchesWithShardedKey\")\n\n\tif err := params.validate(); err != nil {\n\t\tpanic(fmt.Errorf(\"GroupIntoBatchesWithShardedKey: %w\", err))\n\t}\n\tif !typex.IsKV(col.Type()) {\n\t\tpanic(fmt.Errorf(\n\t\t\t\"GroupIntoBatchesWithShardedKey: input PCollection must be KV-typed; got %v\",\n\t\t\tcol.Type()))\n\t}\n\tkeyFT := col.Type().Components()[0]\n\tvar zero K\n\tif keyFT.Type() != reflect.TypeOf(zero) {\n\t\tpanic(fmt.Errorf(\n\t\t\t\"GroupIntoBatchesWithShardedKey: type parameter K (%v) does not match input key type (%v)\",\n\t\t\treflect.TypeOf(zero), keyFT.Type()))\n\t}\n\n\twrapped := beam.ParDo(s, &wrapShardedKeyFn[K]{}, col)\n\treturn GroupIntoBatches(s, params, wrapped)\n}\n","sourceCodeStart":642,"sourceCodeEnd":678,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/transforms/batch/batch.go#L642-L678","documentation":"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.","triggerScenarios":"Calling GroupIntoBatchesWithShardedKey[K](s, params, col) with invalid Params (zero limits, negative BatchSizeBytes, negative MaxBufferingDuration).","commonSituations":"Reusing default-constructed Params{} with no limits; config flags that set both limits to 0 to 'disable' batching.","solutions":["Inspect the wrapped validate error in the panic message","Set BatchSize > 0 or BatchSizeBytes > 0","Ensure no Params fields are negative","Validate Params before building the pipeline"],"exampleFix":"// before\nparams := batch.Params{} // invalid: no limits\n// after\nparams := batch.Params{BatchSizeBytes: 1 << 20, BatchSize: 1000}","handlingStrategy":"validation","validationCode":"if err := params.validate(); err != nil { return fmt.Errorf(\"sharded batching: %w\", err) }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Set explicit positive limits in Params","Validate Params in a shared constructor","Cover Params edge cases in tests"],"tags":["go","apache-beam","panic","invalid-params"],"backgroundTag":"invalid-config-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}