{"record":{"id":"1ca14a7d127f3d10","repo":"apache/beam","slug":"expected-numshards-to-be-provided-when-eos-is-set-to-true","errorCode":null,"errorMessage":"Expected numShards to be provided when EOS is set to true","messagePattern":"Expected numShards to be provided when EOS is set to true","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java","lineNumber":620,"sourceCode":"        if (keySerializerBytes != null) {\n          transform = transform.withKeySerializer((Class) fromByteArray(keySerializerBytes));\n        }\n        byte[] valueSerializerBytes = configRow.getBytes(\"value_serializer\");\n        if (valueSerializerBytes != null) {\n          transform = transform.withValueSerializer((Class) fromByteArray(valueSerializerBytes));\n        }\n        byte[] producerFactoryFnBytes = configRow.getBytes(\"producer_factory_fn\");\n        if (producerFactoryFnBytes != null) {\n          transform =\n              transform.withProducerFactoryFn(\n                  (SerializableFunction) fromByteArray(producerFactoryFnBytes));\n        }\n        Boolean isEOS = configRow.getBoolean(\"eos\");\n        if (isEOS != null && isEOS) {\n          Integer numShards = configRow.getInt32(\"num_shards\");\n          String sinkGroupId = configRow.getString(\"sink_group_id\");\n          if (numShards == null) {\n            throw new IllegalArgumentException(\n                \"Expected numShards to be provided when EOS is set to true\");\n          }\n          if (sinkGroupId == null) {\n            throw new IllegalArgumentException(\n                \"Expected sinkGroupId to be provided when EOS is set to true\");\n          }\n          transform = transform.withEOS(numShards, sinkGroupId);\n        }\n        byte[] consumerFactoryFnBytes = configRow.getBytes(\"consumer_factory_fn\");\n        if (consumerFactoryFnBytes != null) {\n          transform =\n              transform.withConsumerFactoryFn(\n                  (SerializableFunction) fromByteArray(consumerFactoryFnBytes));\n        }\n\n        Map<String, byte[]> producerConfig = configRow.getMap(\"producer_config\");\n        if (producerConfig != null && !producerConfig.isEmpty()) {\n          Map<String, Object> updatedProducerConfig = new HashMap<>();","sourceCodeStart":602,"sourceCodeEnd":638,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java#L602-L638","documentation":"fromConfigRow reconstructs a KafkaIO read/write transform from its config Row. When the 'eos' (exactly-once semantics) flag is true, the sink requires numShards to rebuild withEOS(numShards, sinkGroupId); a missing value is invalid config, so IllegalArgumentException is thrown.","triggerScenarios":"Calling fromConfigRow (via readTransformFromRow) on a config Row where row.getBoolean(\"eos\") is true but row.getInt32(\"num_shards\") is null — i.e., the config was produced/stored without num_shards despite EOS enabled.","commonSituations":"Hand-written or externally generated KafkaIO config rows, config rows produced by an older/different writer that omitted num_shards, or manual edits to serialized pipeline configs that enabled eos without providing shard count.","solutions":["Provide the num_shards field (Integer) in the config Row whenever eos is true","Regenerate the config Row through toConfigRow from a valid KafkaIO write transform instead of hand-building it","Set eos=false if exactly-once semantics are not required, avoiding the num_shards requirement"],"exampleFix":"// before\nRow config = Row.withSchema(schema)\n    .withFieldValues(ImmutableMap.of(\"eos\", true))\n    .build();\n// after\nRow config = Row.withSchema(schema)\n    .withFieldValues(ImmutableMap.of(\"eos\", true, \"num_shards\", 4, \"sink_group_id\", \"my-sink-group\"))\n    .build();","handlingStrategy":"validation","validationCode":"Row config = ...;\nBoolean eos = config.getBoolean(\"eos\");\nif (Boolean.TRUE.equals(eos) && config.getInt32(\"num_shards\") == null) {\n  throw new IllegalArgumentException(\"num_shards is required when eos=true\");\n}","typeGuard":"static boolean hasNumShards(Row row) {\n  return row.getInt32(\"num_shards\") != null;\n}","tryCatchPattern":"try {\n  transform = fromConfigRow(configRow);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().startsWith(\"Expected numShards\")) {\n    // supply default num_shards or fail config validation earlier\n  }\n}","preventionTips":["Always generate KafkaIO config rows via toConfigRow, never hand-build them","Validate config row completeness (eos, num_shards, sink_group_id) at ingestion time","Document required EOS fields in your config templates"],"tags":["java","apache-beam","kafka","config-validation","exactly-once"],"backgroundTag":"missing-required-config-field","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}