apache/beam · error · IllegalArgumentException

Expected numShards to be provided when EOS is set to true

Error message

Expected numShards to be provided when EOS is set to true

What it means

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.

Source

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

        if (keySerializerBytes != null) {
          transform = transform.withKeySerializer((Class) fromByteArray(keySerializerBytes));
        }
        byte[] valueSerializerBytes = configRow.getBytes("value_serializer");
        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<>();

View on GitHub (pinned to 12126d8942)

Solutions

1. Provide the num_shards field (Integer) in the config Row whenever eos is true
2. Regenerate the config Row through toConfigRow from a valid KafkaIO write transform instead of hand-building it
3. Set eos=false if exactly-once semantics are not required, avoiding the num_shards requirement

Example fix

// before
Row config = Row.withSchema(schema)
    .withFieldValues(ImmutableMap.of("eos", true))
    .build();
// after
Row config = Row.withSchema(schema)
    .withFieldValues(ImmutableMap.of("eos", true, "num_shards", 4, "sink_group_id", "my-sink-group"))
    .build();

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

static boolean hasNumShards(Row row) {
  return row.getInt32("num_shards") != null;
}

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().startsWith("Expected numShards")) {
    // supply default num_shards or fail config validation earlier
  }
}

When it happens

Trigger: 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.

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

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 sinkGroupId 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/1ca14a7d127f3d10. Report an issue: GitHub.