{"record":{"id":"480c7dc8f80ca8da","repo":"apache/beam","slug":"unsupported-watermark-policy","errorCode":null,"errorMessage":"Unsupported watermark_policy: ","messagePattern":"Unsupported watermark_policy: ","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisReadSchemaTransformProvider.java","lineNumber":278,"sourceCode":"      }\n      if (config.getMaxCapacityPerShard() != null) {\n        readTransform =\n            readTransform.withMaxCapacityPerShard(config.getMaxCapacityPerShard().intValue());\n      }\n      if (config.getWatermarkPolicy() != null) {\n        switch (config.getWatermarkPolicy()) {\n          case \"ARRIVAL_TIME\":\n            readTransform =\n                config.getWatermarkIdleDurationThreshold() != null\n                    ? readTransform.withArrivalTimeWatermarkPolicy(\n                        Duration.millis(config.getWatermarkIdleDurationThreshold()))\n                    : readTransform.withArrivalTimeWatermarkPolicy();\n            break;\n          case \"PROCESSING_TIME\":\n            readTransform = readTransform.withProcessingTimeWatermarkPolicy();\n            break;\n          default:\n            throw new IllegalArgumentException(\n                \"Unsupported watermark_policy: \" + config.getWatermarkPolicy());\n        }\n      }\n      if (config.getRateLimit() != null) {\n        readTransform =\n            readTransform.withFixedDelayRateLimitPolicy(Duration.millis(config.getRateLimit()));\n      }\n\n      PCollection<Row> output =\n          input\n              .getPipeline()\n              .apply(readTransform)\n              .apply(\"KinesisRecordToRow\", ParDo.of(new KinesisRecordToRowFn()))\n              .setRowSchema(OUTPUT_SCHEMA);\n\n      return PCollectionRowTuple.of(OUTPUT_TAG, output);\n    }\n  }","sourceCodeStart":260,"sourceCodeEnd":296,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisReadSchemaTransformProvider.java#L260-L296","documentation":"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.","triggerScenarios":"Supplying watermark_policy in the schema-transform config with a misspelled, lowercased, or unsupported value such as 'processingtime', 'event_time', or 'none'.","commonSituations":"YAML config typos; following older documentation that used different policy names; translating from another connector's watermark option names.","solutions":["Set watermark_policy to one of the supported values: ARRIVAL_TIME, PROCESSING_TIME, or EMBEDDED (exact uppercase spelling)","Check the provider's class source/docs for the exact accepted strings on your Beam version","Remove the watermark_policy key to use the default policy"],"exampleFix":"// before\nwatermark_policy: processingtime\n// after\nwatermark_policy: PROCESSING_TIME","handlingStrategy":"validation","validationCode":"Set<String> ok = Set.of(\"ARRIVAL_TIME\",\"PROCESSING_TIME\",\"EMBEDDED\");\nif (cfg.getWatermarkPolicy()!=null && !ok.contains(cfg.getWatermarkPolicy())) throw new IllegalArgumentException(\"bad watermark_policy\");","typeGuard":"boolean validWatermarkPolicy(String s){ return s==null || Set.of(\"ARRIVAL_TIME\",\"PROCESSING_TIME\",\"EMBEDDED\").contains(s); }","tryCatchPattern":"try { expand(cfg); } catch (IllegalArgumentException e) { correct watermark_policy spelling/case in YAML; resubmit; }","preventionTips":["Use exact uppercase enum-style names","Check the provider source for accepted values on your Beam version","Validate YAML configs against the provider's documented schema"],"tags":["kinesis","watermark","invalid-enum","config"],"backgroundTag":"invalid-enum-value","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"}