apache/beam · error · ValueError

timestamp_policy should be one of [ProcessingTime…

Error message

timestamp_policy should be one of [ProcessingTime, CreateTime, LogAppendTime]

What it means

ReadFromKafka's constructor validates timestamp_policy against the three supported policies: PROCESSING_TIME ('processing_time'), CREATE_TIME ('create_time'), and LOG_APPEND_TIME ('log_append_time'). Any other string raises ValueError. The policy determines how record timestamps are assigned in the produced elements.

Solutions

Use one of ReadFromKafka.create_time_policy, .processing_time_policy, or .log_append_time constants.
If passing a raw string, use exactly 'processing_time', 'create_time', or 'log_append_time' (lowercase).
Omit the argument to use the default policy.

Example fix

# before
ReadFromKafka(consumer_config, ['topic'], timestamp_policy='CreateTime')
# after
ReadFromKafka(consumer_config, ['topic'], timestamp_policy=ReadFromKafka.create_time_policy)

Defensive patterns

Strategy: validation

Validate before calling

VALID = (ReadFromKafka.processing_time_policy, ReadFromKafka.create_time_policy, ReadFromKafka.log_append_time)
assert timestamp_policy in VALID, timestamp_policy

Type guard

def is_valid_timestamp_policy(policy) -> bool:
    return policy in (ReadFromKafka.processing_time_policy,
                      ReadFromKafka.create_time_policy,
                      ReadFromKafka.log_append_time)

Try / catch

try:
    _ = ReadFromKafka(consumer_config, topics, timestamp_policy=policy)
except ValueError as e:
    if 'timestamp_policy' in str(e): log.error('Invalid policy %r; use class constants', policy)

Prevention

Use ReadFromKafka class constants instead of raw strings
Remember values are lowercase ('create_time', not 'CreateTime')
Validate config-file-driven policy values at load time

When it happens

Trigger: Calling ReadFromKafka(..., timestamp_policy='CreateTime') or 'CREATED_TIME' or any misspelled/uppercase value not matching the class constants ReadFromKafka.processing_time_policy / create_time_policy / log_append_time.

Common situations: Hand-typing the policy string with wrong casing or wording instead of using the class constants; copying a timestamp_policy name from the Java SDK.

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/5298b67ee7b301c9. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/io/kafka.py:232

    :param redistribute_num_keys: Configures how many keys the Redistribute
        spreads the data across.
    :param allow_duplicates: whether the Redistribute transform allows for
        duplicates (this serves solely as a hint to the underlying runner).
    :param dynamic_read_poll_interval_seconds: The interval in seconds at which
        to check for new partitions. If not None, dynamic partition discovery
        is enabled.
    :param consumer_factory_fn_class: A fully qualified classpath to an
        existing provided consumerFactoryFn. If not None, this will construct
        Kafka consumers with a custom configuration.
    :param consumer_factory_fn_params: A map which specifies the parameters for
        the provided consumer_factory_fn_class. If not None, the values in this
        map will be used when constructing the consumer_factory_fn_class object.
        This cannot be null if the consumer_factory_fn_class is not null.
    """
    if timestamp_policy not in [ReadFromKafka.processing_time_policy,
                                ReadFromKafka.create_time_policy,
                                ReadFromKafka.log_append_time]:
      raise ValueError(
          'timestamp_policy should be one of '
          '[ProcessingTime, CreateTime, LogAppendTime]')

    super().__init__(
        self.URN_WITH_METADATA if with_metadata else self.URN_WITHOUT_METADATA,
        NamedTupleBasedPayloadBuilder(
            ReadFromKafkaSchema(
                consumer_config=consumer_config,
                topics=topics,
                key_deserializer=key_deserializer,
                value_deserializer=value_deserializer,
                max_num_records=max_num_records,
                max_read_time=max_read_time,
                start_read_time=start_read_time,
                commit_offset_in_finalize=commit_offset_in_finalize,
                timestamp_policy=timestamp_policy,
                consumer_polling_timeout=consumer_polling_timeout,
                redistribute=redistribute,

View on GitHub (pinned to 12126d8942)