{"record":{"id":"5298b67ee7b301c9","repo":"apache/beam","slug":"timestamp-policy-should-be-one-of-processingtime-createtime","errorCode":null,"errorMessage":"timestamp_policy should be one of [ProcessingTime, CreateTime, LogAppendTime]","messagePattern":"timestamp_policy should be one of \\[ProcessingTime, CreateTime, LogAppendTime\\]","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/kafka.py","lineNumber":232,"sourceCode":"    :param redistribute_num_keys: Configures how many keys the Redistribute\n        spreads the data across.\n    :param allow_duplicates: whether the Redistribute transform allows for\n        duplicates (this serves solely as a hint to the underlying runner).\n    :param dynamic_read_poll_interval_seconds: The interval in seconds at which\n        to check for new partitions. If not None, dynamic partition discovery\n        is enabled.\n    :param consumer_factory_fn_class: A fully qualified classpath to an\n        existing provided consumerFactoryFn. If not None, this will construct\n        Kafka consumers with a custom configuration.\n    :param consumer_factory_fn_params: A map which specifies the parameters for\n        the provided consumer_factory_fn_class. If not None, the values in this\n        map will be used when constructing the consumer_factory_fn_class object.\n        This cannot be null if the consumer_factory_fn_class is not null.\n    \"\"\"\n    if timestamp_policy not in [ReadFromKafka.processing_time_policy,\n                                ReadFromKafka.create_time_policy,\n                                ReadFromKafka.log_append_time]:\n      raise ValueError(\n          'timestamp_policy should be one of '\n          '[ProcessingTime, CreateTime, LogAppendTime]')\n\n    super().__init__(\n        self.URN_WITH_METADATA if with_metadata else self.URN_WITHOUT_METADATA,\n        NamedTupleBasedPayloadBuilder(\n            ReadFromKafkaSchema(\n                consumer_config=consumer_config,\n                topics=topics,\n                key_deserializer=key_deserializer,\n                value_deserializer=value_deserializer,\n                max_num_records=max_num_records,\n                max_read_time=max_read_time,\n                start_read_time=start_read_time,\n                commit_offset_in_finalize=commit_offset_in_finalize,\n                timestamp_policy=timestamp_policy,\n                consumer_polling_timeout=consumer_polling_timeout,\n                redistribute=redistribute,","sourceCodeStart":214,"sourceCodeEnd":250,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/kafka.py#L214-L250","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"# before\nReadFromKafka(consumer_config, ['topic'], timestamp_policy='CreateTime')\n# after\nReadFromKafka(consumer_config, ['topic'], timestamp_policy=ReadFromKafka.create_time_policy)","handlingStrategy":"validation","validationCode":"VALID = (ReadFromKafka.processing_time_policy, ReadFromKafka.create_time_policy, ReadFromKafka.log_append_time)\nassert timestamp_policy in VALID, timestamp_policy","typeGuard":"def is_valid_timestamp_policy(policy) -> bool:\n    return policy in (ReadFromKafka.processing_time_policy,\n                      ReadFromKafka.create_time_policy,\n                      ReadFromKafka.log_append_time)","tryCatchPattern":"try:\n    _ = ReadFromKafka(consumer_config, topics, timestamp_policy=policy)\nexcept ValueError as e:\n    if 'timestamp_policy' in str(e): log.error('Invalid policy %r; use class constants', policy)","preventionTips":["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"],"tags":["apache-beam","python","kafka","invalid-enum-value","timestamp-policy"],"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"}