apache/beam · error · IllegalArgumentException

Expected watchTopicPartitionDuration to be available when…

Error message

Expected watchTopicPartitionDuration to be available when isDynamicRead is set to true

What it means

If the config Row marks is_dynamic_read=true, the watch_topic_partition_duration value must also be present, because withDynamicRead requires a refresh duration. A Row with dynamic read enabled but no duration is inconsistent and rejected with IllegalArgumentException.

Solutions

1. Populate watch_topic_partition_duration in the config Row (a Duration value).
2. Or set is_dynamic_read to false if dynamic partition discovery is not intended.
3. Regenerate the payload from a transform built with withDynamicRead(Duration) via row().

Example fix

// before
.withDynamicRead() // no duration captured
// after
.withDynamicRead(Duration.standardMinutes(5))

Boolean dyn = configRow.getBoolean("is_dynamic_read");
if (dyn != null && dyn && configRow.getValue("watch_topic_partition_duration") == null) {
  throw new IllegalArgumentException("is_dynamic_read=true requires watch_topic_partition_duration");
}

boolean dynamicReadConsistent(Row r) {
  Boolean d = r.getBoolean("is_dynamic_read");
  return d == null || !d || r.getValue("watch_topic_partition_duration") != null;
}

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("watchTopicPartitionDuration")) { /* add duration or disable dynamic read */ }
  throw e;
}

When it happens

Trigger: Restoring a KafkaIO read transform from a config Row where is_dynamic_read is true but watch_topic_partition_duration is null (e.g. transform was built via withDynamicRead without duration, or the Row field was dropped).

Common situations: Partially serialized payloads from older Beam versions predating the duration field; hand-edited config Rows enabling dynamic reads; mismatched field population in custom translation tooling.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

• Encoded value of the consumer config property
• External environment endpoint must be set.
• Unable to parse jar URL
• Allow list file does not exist
• Ambiguous expression type (perhaps missing quoting?)

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

Appendix: source

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

        Instant startReadTime = configRow.getValue("start_read_time");
        if (startReadTime != null) {
          transform = transform.withStartReadTime(startReadTime);
        }
        Instant stopReadTime = configRow.getValue("stop_read_time");
        if (stopReadTime != null) {
          transform = transform.withStopReadTime(stopReadTime);
        }
        Boolean isCommitOffsetFinalizeEnabled =
            configRow.getBoolean("is_commit_offset_finalize_enabled");
        if (isCommitOffsetFinalizeEnabled != null && isCommitOffsetFinalizeEnabled) {
          transform = transform.commitOffsetsInFinalize();
        }
        Boolean isDynamicRead = configRow.getBoolean("is_dynamic_read");
        if (isDynamicRead != null && isDynamicRead) {
          Duration watchTopicPartitionDuration =
              configRow.getValue("watch_topic_partition_duration");
          if (watchTopicPartitionDuration == null) {
            throw new IllegalArgumentException(
                "Expected watchTopicPartitionDuration to be available when isDynamicRead is set to true");
          }
          transform =
              transform.withDynamicRead(
                  org.joda.time.Duration.millis(watchTopicPartitionDuration.toMillis()));
        }

        byte[] timestampPolicyFactory = configRow.getBytes("timestamp_policy_factory");
        if (timestampPolicyFactory != null) {
          transform =
              transform.withTimestampPolicyFactory(
                  (TimestampPolicyFactory) fromByteArray(timestampPolicyFactory));
        }
        Map<String, byte[]> offsetConsumerConfig = configRow.getMap("offset_consumer_config");
        if (offsetConsumerConfig != null) {
          Map<String, Object> updatedOffsetConsumerConfig = new HashMap<>();
          offsetConsumerConfig.forEach(
              (key, dataBytes) -> {

View on GitHub (pinned to 12126d8942)