{"record":{"id":"33fc94b41d69f8ce","repo":"apache/beam","slug":"expected-watchtopicpartitionduration-to-be-available-when","errorCode":null,"errorMessage":"Expected watchTopicPartitionDuration to be available when isDynamicRead is set to true","messagePattern":"Expected watchTopicPartitionDuration to be available when isDynamicRead 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":408,"sourceCode":"        Instant startReadTime = configRow.getValue(\"start_read_time\");\n        if (startReadTime != null) {\n          transform = transform.withStartReadTime(startReadTime);\n        }\n        Instant stopReadTime = configRow.getValue(\"stop_read_time\");\n        if (stopReadTime != null) {\n          transform = transform.withStopReadTime(stopReadTime);\n        }\n        Boolean isCommitOffsetFinalizeEnabled =\n            configRow.getBoolean(\"is_commit_offset_finalize_enabled\");\n        if (isCommitOffsetFinalizeEnabled != null && isCommitOffsetFinalizeEnabled) {\n          transform = transform.commitOffsetsInFinalize();\n        }\n        Boolean isDynamicRead = configRow.getBoolean(\"is_dynamic_read\");\n        if (isDynamicRead != null && isDynamicRead) {\n          Duration watchTopicPartitionDuration =\n              configRow.getValue(\"watch_topic_partition_duration\");\n          if (watchTopicPartitionDuration == null) {\n            throw new IllegalArgumentException(\n                \"Expected watchTopicPartitionDuration to be available when isDynamicRead is set to true\");\n          }\n          transform =\n              transform.withDynamicRead(\n                  org.joda.time.Duration.millis(watchTopicPartitionDuration.toMillis()));\n        }\n\n        byte[] timestampPolicyFactory = configRow.getBytes(\"timestamp_policy_factory\");\n        if (timestampPolicyFactory != null) {\n          transform =\n              transform.withTimestampPolicyFactory(\n                  (TimestampPolicyFactory) fromByteArray(timestampPolicyFactory));\n        }\n        Map<String, byte[]> offsetConsumerConfig = configRow.getMap(\"offset_consumer_config\");\n        if (offsetConsumerConfig != null) {\n          Map<String, Object> updatedOffsetConsumerConfig = new HashMap<>();\n          offsetConsumerConfig.forEach(\n              (key, dataBytes) -> {","sourceCodeStart":390,"sourceCodeEnd":426,"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#L390-L426","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Populate watch_topic_partition_duration in the config Row (a Duration value).","Or set is_dynamic_read to false if dynamic partition discovery is not intended.","Regenerate the payload from a transform built with withDynamicRead(Duration) via row()."],"exampleFix":"// before\n.withDynamicRead() // no duration captured\n// after\n.withDynamicRead(Duration.standardMinutes(5))\n","handlingStrategy":"validation","validationCode":"Boolean dyn = configRow.getBoolean(\"is_dynamic_read\");\nif (dyn != null && dyn && configRow.getValue(\"watch_topic_partition_duration\") == null) {\n  throw new IllegalArgumentException(\"is_dynamic_read=true requires watch_topic_partition_duration\");\n}","typeGuard":"boolean dynamicReadConsistent(Row r) {\n  Boolean d = r.getBoolean(\"is_dynamic_read\");\n  return d == null || !d || r.getValue(\"watch_topic_partition_duration\") != null;\n}","tryCatchPattern":"try {\n  transform = fromConfigRow(configRow);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"watchTopicPartitionDuration\")) { /* add duration or disable dynamic read */ }\n  throw e;\n}","preventionTips":["Always call withDynamicRead(Duration) with an explicit duration.","Keep is_dynamic_read and watch_topic_partition_duration in sync in custom payload tooling.","Verify upgrade payloads after any Beam version change."],"tags":["kafka","dynamic-read","config","missing-field"],"backgroundTag":"conflicting-config-options","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}