apache/beam · error · IllegalArgumentException

Expected sinkGroupId to be provided when EOS is set to true

Error message

Expected sinkGroupId to be provided when EOS is set to true

What it means

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.

Source

Thrown at sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java:624

        if (valueSerializerBytes != null) {
          transform = transform.withValueSerializer((Class) fromByteArray(valueSerializerBytes));
        }
        byte[] producerFactoryFnBytes = configRow.getBytes("producer_factory_fn");
        if (producerFactoryFnBytes != null) {
          transform =
              transform.withProducerFactoryFn(
                  (SerializableFunction) fromByteArray(producerFactoryFnBytes));
        }
        Boolean isEOS = configRow.getBoolean("eos");
        if (isEOS != null && isEOS) {
          Integer numShards = configRow.getInt32("num_shards");
          String sinkGroupId = configRow.getString("sink_group_id");
          if (numShards == null) {
            throw new IllegalArgumentException(
                "Expected numShards to be provided when EOS is set to true");
          }
          if (sinkGroupId == null) {
            throw new IllegalArgumentException(
                "Expected sinkGroupId to be provided when EOS is set to true");
          }
          transform = transform.withEOS(numShards, sinkGroupId);
        }
        byte[] consumerFactoryFnBytes = configRow.getBytes("consumer_factory_fn");
        if (consumerFactoryFnBytes != null) {
          transform =
              transform.withConsumerFactoryFn(
                  (SerializableFunction) fromByteArray(consumerFactoryFnBytes));
        }

        Map<String, byte[]> producerConfig = configRow.getMap("producer_config");
        if (producerConfig != null && !producerConfig.isEmpty()) {
          Map<String, Object> updatedProducerConfig = new HashMap<>();
          producerConfig.forEach(
              (key, dataBytes) -> {
                try {
                  updatedProducerConfig.put(key, fromByteArray((byte[]) dataBytes));

View on GitHub (pinned to 12126d8942)

Solutions

1. Add the sink_group_id string field to the config Row when eos is true
2. Regenerate the config via toConfigRow from a transform built with withEOS(numShards, sinkGroupId)
3. Disable eos if the sink group id is unknown or not needed

Example fix

// before
ImmutableMap.of("eos", true, "num_shards", 4)
// after
ImmutableMap.of("eos", true, "num_shards", 4, "sink_group_id", "kafka-eos-sink-group")

Row config = ...;
Boolean eos = config.getBoolean("eos");
if (Boolean.TRUE.equals(eos) && config.getString("sink_group_id") == null) {
  throw new IllegalArgumentException("sink_group_id is required when eos=true");
}

static boolean hasSinkGroupId(Row row) {
  return row.getString("sink_group_id") != null;
}

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().startsWith("Expected sinkGroupId")) {
    // backfill sink_group_id or disable eos
  }
}

When it happens

Trigger: Calling fromConfigRow (via readTransformFromRow) on a config Row where row.getBoolean("eos") is true but row.getString("sink_group_id") is null.

Common situations: 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.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Expected numShards to be provided when EOS is set to true
• Unknown watermark type: ${type}. Supported types are Process
• '${fieldName}' field is invalid at the top level for Kafka i
• host must not be empty.
• port must be between 1 and 65535, but was .

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