apache/beam · error · IllegalArgumentException

timestampPolicy should be one of (ProcessingTime…

Error message

timestampPolicy should be one of (ProcessingTime, CreateTime, LogAppendTime)

What it means

KafkaIO's external/cross-language builder (setupExternalBuilder) validates the timestampPolicy string against the three supported values. Any other string cannot be mapped to a TimestampPolicyFactory, so an IllegalArgumentException is thrown.

Solutions

1. Set timestampPolicy to one of the exact strings: ProcessingTime, CreateTime, or LogAppendTime.
2. Fix casing — the comparison is case-sensitive.
3. Omit the field to use the default policy.
4. Validate user-supplied policy strings before submitting the pipeline.

Example fix

// before
{"timestampPolicy": "processingTime"}
// after
{"timestampPolicy": "ProcessingTime"}

if (!Set.of("ProcessingTime","CreateTime","LogAppendTime").contains(policy)) throw new IllegalArgumentException(policy);

try { build(cfg); } catch (IllegalArgumentException e) { fixPolicy(); }

When it happens

Trigger: Passing timestampPolicy in the external KafkaIO configuration (e.g. via SQL/cross-language config) with a value other than exactly 'ProcessingTime', 'CreateTime', or 'LogAppendTime' (case-sensitive).

Common situations: Typo such as 'processingTime' or 'CREATETIME'; passing 'CreateTime' variants like 'create_time'; configuring via template/SQL where the value comes from user input.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• consumerPollingTimeout should be > 0.
• Unable to construct FactoryFn
• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• Couldn't infer Coder from

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:906

        if (config.maxReadTime != null) {
          builder.setMaxReadTime(Duration.standardSeconds(config.maxReadTime));
        }
        builder.setMaxNumRecords(
            config.maxNumRecords == null ? Long.MAX_VALUE : config.maxNumRecords);

        // Set committing offset configuration.
        builder.setCommitOffsetsInFinalizeEnabled(config.commitOffsetInFinalize);

        // Set timestamp policy with built-in types.
        String timestampPolicy = config.timestampPolicy;
        if (timestampPolicy.equals("ProcessingTime")) {
          builder.setTimestampPolicyFactory(TimestampPolicyFactory.withProcessingTime());
        } else if (timestampPolicy.equals("CreateTime")) {
          builder.setTimestampPolicyFactory(TimestampPolicyFactory.withCreateTime(Duration.ZERO));
        } else if (timestampPolicy.equals("LogAppendTime")) {
          builder.setTimestampPolicyFactory(TimestampPolicyFactory.withLogAppendTime());
        } else {
          throw new IllegalArgumentException(
              "timestampPolicy should be one of (ProcessingTime, CreateTime, LogAppendTime)");
        }

        if (config.startReadTime != null) {
          builder.setStartReadTime(Instant.ofEpochMilli(config.startReadTime));
        }

        if (config.stopReadTime != null) {
          builder.setStopReadTime(Instant.ofEpochMilli(config.stopReadTime));
        }

        if (config.dynamicReadPollIntervalSeconds != null) {
          builder.setDynamicRead(true);
          builder.setWatchTopicPartitionDuration(
              Duration.standardSeconds(config.dynamicReadPollIntervalSeconds));
        } else {
          builder.setDynamicRead(false);
        }

View on GitHub (pinned to 12126d8942)