apache/beam · error · IllegalArgumentException

Unsupported watermark_policy:

Error message

Unsupported watermark_policy:

What it means

KinesisReadSchemaTransformProvider.expand maps config.getWatermarkPolicy() (a string) onto one of KinesisIO's watermark policies. Only a fixed set (e.g. ARRIVAL_TIME, PROCESSING_TIME, EMBEDDED) is accepted; anything else hits the default branch and throws IllegalArgumentException.

Solutions

  1. Set watermark_policy to one of the supported values: ARRIVAL_TIME, PROCESSING_TIME, or EMBEDDED (exact uppercase spelling)
  2. Check the provider's class source/docs for the exact accepted strings on your Beam version
  3. Remove the watermark_policy key to use the default policy

Example fix

// before
watermark_policy: processingtime
// after
watermark_policy: PROCESSING_TIME
Defensive patterns

Strategy: validation

Validate before calling

Set<String> ok = Set.of("ARRIVAL_TIME","PROCESSING_TIME","EMBEDDED");
if (cfg.getWatermarkPolicy()!=null && !ok.contains(cfg.getWatermarkPolicy())) throw new IllegalArgumentException("bad watermark_policy");

Type guard

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

Try / catch

try { expand(cfg); } catch (IllegalArgumentException e) { correct watermark_policy spelling/case in YAML; resubmit; }

Prevention

When it happens

Trigger: Supplying watermark_policy in the schema-transform config with a misspelled, lowercased, or unsupported value such as 'processingtime', 'event_time', or 'none'.

Common situations: YAML config typos; following older documentation that used different policy names; translating from another connector's watermark option names.

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

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

Appendix: source

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

      }
      if (config.getMaxCapacityPerShard() != null) {
        readTransform =
            readTransform.withMaxCapacityPerShard(config.getMaxCapacityPerShard().intValue());
      }
      if (config.getWatermarkPolicy() != null) {
        switch (config.getWatermarkPolicy()) {
          case "ARRIVAL_TIME":
            readTransform =
                config.getWatermarkIdleDurationThreshold() != null
                    ? readTransform.withArrivalTimeWatermarkPolicy(
                        Duration.millis(config.getWatermarkIdleDurationThreshold()))
                    : readTransform.withArrivalTimeWatermarkPolicy();
            break;
          case "PROCESSING_TIME":
            readTransform = readTransform.withProcessingTimeWatermarkPolicy();
            break;
          default:
            throw new IllegalArgumentException(
                "Unsupported watermark_policy: " + config.getWatermarkPolicy());
        }
      }
      if (config.getRateLimit() != null) {
        readTransform =
            readTransform.withFixedDelayRateLimitPolicy(Duration.millis(config.getRateLimit()));
      }

      PCollection<Row> output =
          input
              .getPipeline()
              .apply(readTransform)
              .apply("KinesisRecordToRow", ParDo.of(new KinesisRecordToRowFn()))
              .setRowSchema(OUTPUT_SCHEMA);

      return PCollectionRowTuple.of(OUTPUT_TAG, output);
    }
  }

View on GitHub (pinned to 12126d8942)