apache/beam · error · RuntimeError

Invalid watermark policy

Error message

Invalid watermark policy: {}

What it means

WatermarkPolicy.validate_param checks that a given watermark policy name is a valid class attribute of WatermarkPolicy (e.g. 'ARRIVAL_TIME' or 'PROCESSING_TYPE'). When a truthy string is passed that does not match any defined policy, a RuntimeError is raised with the offending value. This fails fast at pipeline construction time instead of producing a confusing failure inside the Kinesis reader later.

Solutions

1. Use one of the defined constants: apache_beam.io.kinesis.WatermarkPolicy.ARRIVAL_TIME or WatermarkPolicy.PROCESSING_TYPE, instead of a raw string.
2. Check the exact spelling and case of the policy string; the check is hasattr-based and case-sensitive.
3. Pass None or omit the parameter if you do not need watermark updates, since empty values are accepted.
4. Inspect the WatermarkPolicy class in your installed Beam version (sdks/python/apache_beam/io/kinesis.py) to see which values are supported.

Example fix

// before
ReadDataFromKinesis(input_stream='stream', watermark_policy='Arrival_Time')
// after
from apache_beam.io.kinesis import WatermarkPolicy, ReadDataFromKinesis
ReadDataFromKinesis(input_stream='stream', watermark_policy=WatermarkPolicy.ARRIVAL_TIME)

from apache_beam.io.kinesis import WatermarkPolicy

def valid_watermark_policy(p):
    return not p or hasattr(WatermarkPolicy, p)

assert valid_watermark_policy(watermark_policy), f"unknown policy: {watermark_policy}"

def is_watermark_policy(value) -> bool:
    from apache_beam.io.kinesis import WatermarkPolicy
    return isinstance(value, str) and hasattr(WatermarkPolicy, value)

When it happens

Trigger: Calling ReadDataFromKinesis/beam.io.kinesis.ReadFromKinesis with watermark_policy set to any string not present as an attribute on WatermarkPolicy, e.g. watermark_policy='Arrival_Time', 'arrival_time' or a typo like 'ARRIVALL_TIME'. Empty/None passes validation because `if param` short-circuits.

Common situations: Hand-typing the policy name in pipeline options or YAML templates instead of referencing WatermarkPolicy.ARRIVAL_TIME / WatermarkPolicy.PROCESSING_TYPE; case-sensitivity mistakes after copying config between jobs; renaming/upgrade where the allowed policy values changed.

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

• Invalid initial position in stream
• bucket_boundaries must be a non-empty sorted list of…
• Cannot create object with unspecified or no compression
• ignore_first_n_seen_per_batch_size
• Minimum ( ) must not be greater than maximum ( )

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

Appendix: source

Thrown at sdks/python/apache_beam/io/kinesis.py:347

class InitialPositionInStream:
  LATEST = 'LATEST'
  TRIM_HORIZON = 'TRIM_HORIZON'
  AT_TIMESTAMP = 'AT_TIMESTAMP'

  @staticmethod
  def validate_param(param):
    if param and not hasattr(InitialPositionInStream, param):
      raise RuntimeError('Invalid initial position in stream: {}'.format(param))


class WatermarkPolicy:
  PROCESSING_TYPE = 'PROCESSING_TYPE'
  ARRIVAL_TIME = 'ARRIVAL_TIME'

  @staticmethod
  def validate_param(param):
    if param and not hasattr(WatermarkPolicy, param):
      raise RuntimeError('Invalid watermark policy: {}'.format(param))

View on GitHub (pinned to 12126d8942)