{"record":{"id":"bc7bca98be05e815","repo":"apache/beam","slug":"expected-sinkgroupid-to-be-provided-when-eos-is-set-to-true","errorCode":null,"errorMessage":"Expected sinkGroupId to be provided when EOS is set to true","messagePattern":"Expected sinkGroupId 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":624,"sourceCode":"        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<>();\n          producerConfig.forEach(\n              (key, dataBytes) -> {\n                try {\n                  updatedProducerConfig.put(key, fromByteArray((byte[]) dataBytes));","sourceCodeStart":606,"sourceCodeEnd":642,"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#L606-L642","documentation":"fromConfigRow requires sinkGroupId when 'eos' is true because reconstructing the sink calls withEOS(numShards, sinkGroupId); Kafka exactly-once semantics depend on a stable sink consumer group id. A null sink_group_id makes the config invalid and throws IllegalArgumentException.","triggerScenarios":"Calling fromConfigRow (via readTransformFromRow) on a config Row where row.getBoolean(\"eos\") is true but row.getString(\"sink_group_id\") is null.","commonSituations":"Manually assembled KafkaIO config rows, configs migrated between systems losing the sink_group_id field, or pipelines that enabled EOS without ever setting a sink group id.","solutions":["Add the sink_group_id string field to the config Row when eos is true","Regenerate the config via toConfigRow from a transform built with withEOS(numShards, sinkGroupId)","Disable eos if the sink group id is unknown or not needed"],"exampleFix":"// before\nImmutableMap.of(\"eos\", true, \"num_shards\", 4)\n// after\nImmutableMap.of(\"eos\", true, \"num_shards\", 4, \"sink_group_id\", \"kafka-eos-sink-group\")","handlingStrategy":"validation","validationCode":"Row config = ...;\nBoolean eos = config.getBoolean(\"eos\");\nif (Boolean.TRUE.equals(eos) && config.getString(\"sink_group_id\") == null) {\n  throw new IllegalArgumentException(\"sink_group_id is required when eos=true\");\n}","typeGuard":"static boolean hasSinkGroupId(Row row) {\n  return row.getString(\"sink_group_id\") != null;\n}","tryCatchPattern":"try {\n  transform = fromConfigRow(configRow);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().startsWith(\"Expected sinkGroupId\")) {\n    // backfill sink_group_id or disable eos\n  }\n}","preventionTips":["Pair every eos=true config with a stable sink_group_id value","Round-trip test configs through toConfigRow/fromConfigRow in CI","Don't manually edit serialized pipeline configs without re-validating"],"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"}