apache/beam · error · RuntimeException

Unsupported watermark policy type

Error message

Unsupported watermark policy type: %s

What it means

The registrar translates configuration.watermarkPolicy (enum-like) onto KinesisIO watermark policies. Values outside the known set (ARRIVAL_TIME, PROCESSING_TIME, EMBEDDED) fall into the default branch and throw a RuntimeException naming the unsupported policy. Unlike the SchemaTransform provider, this path uses an enum/constant, so only exact matches compile-time-known values pass.

Solutions

1. Use exactly ARRIVAL_TIME or PROCESSING_TIME (or the EMBEDDED option if supported) in configuration.watermarkPolicy
2. Match the enum constant names defined in KinesisTransformRegistrar/KinesisIO for your Beam version
3. Omit watermarkPolicy to use the default

Example fix

// before
"watermarkPolicy": "EVENT_TIME"
// after
"watermarkPolicy": "ARRIVAL_TIME"

if (cfg.watermarkPolicy != null && !Set.of("ARRIVAL_TIME","PROCESSING_TIME","EMBEDDED").contains(cfg.watermarkPolicy)) throw new IllegalArgumentException("unsupported watermarkPolicy");

boolean knownPolicy(String s){ return s==null || Set.of("ARRIVAL_TIME","PROCESSING_TIME","EMBEDDED").contains(s); }

try { buildExternal(cfg); } catch (RuntimeException e) { correct watermarkPolicy to a supported constant; resubmit; }

When it happens

Trigger: Passing a watermarkPolicy string that doesn't match any configured enum constant in the cross-language pipeline spec, e.g. 'EVENT_TIME' or 'arrivaltime'.

Common situations: Copy-pasted options from the SchemaTransform provider (string-based) into cross-language JSON config with different casing; drift between Beam versions' supported policies.

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

• Unsupported watermark_policy:
• Service endpoint must be URI format, got
• Ambiguous renaming of tags.
• attempted to add namespace to missing coder id
• attempted to add namespace to missing windowing strategy id

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

Appendix: source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisTransformRegistrar.java:303

      }
      if (configuration.maxCapacityPerShard != null) {
        readTransform =
            readTransform.withMaxCapacityPerShard(configuration.maxCapacityPerShard.intValue());
      }
      if (configuration.watermarkPolicy != null) {
        switch (configuration.watermarkPolicy) {
          case ARRIVAL_TIME:
            readTransform =
                configuration.watermarkIdleDurationThreshold != null
                    ? readTransform.withArrivalTimeWatermarkPolicy(
                        configuration.watermarkIdleDurationThreshold)
                    : readTransform.withArrivalTimeWatermarkPolicy();
            break;
          case PROCESSING_TIME:
            readTransform = readTransform.withProcessingTimeWatermarkPolicy();
            break;
          default:
            throw new RuntimeException(
                String.format(
                    "Unsupported watermark policy type: %s", configuration.watermarkPolicy));
        }
      }
      if (configuration.rateLimit != null) {
        readTransform = readTransform.withFixedDelayRateLimitPolicy(configuration.rateLimit);
      }
      if (configuration.maxReadTime != null) {
        readTransform = readTransform.withMaxReadTime(configuration.maxReadTime);
      }
      if (configuration.initialPositionInStream != null) {
        readTransform =
            readTransform.withInitialPositionInStream(configuration.initialPositionInStream);
      }
      if (configuration.requestRecordsLimit != null) {
        readTransform = readTransform.withRequestRecordsLimit(configuration.requestRecordsLimit);
      }
      if (configuration.initialTimestampInStream != null) {

View on GitHub (pinned to 12126d8942)