{"record":{"id":"54de4100a4174e14","repo":"apache/beam","slug":"timestamppolicy-should-be-one-of-processingtime-createtime","errorCode":null,"errorMessage":"timestampPolicy should be one of (ProcessingTime, CreateTime, LogAppendTime)","messagePattern":"timestampPolicy should be one of \\(ProcessingTime, CreateTime, LogAppendTime\\)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java","lineNumber":906,"sourceCode":"        if (config.maxReadTime != null) {\n          builder.setMaxReadTime(Duration.standardSeconds(config.maxReadTime));\n        }\n        builder.setMaxNumRecords(\n            config.maxNumRecords == null ? Long.MAX_VALUE : config.maxNumRecords);\n\n        // Set committing offset configuration.\n        builder.setCommitOffsetsInFinalizeEnabled(config.commitOffsetInFinalize);\n\n        // Set timestamp policy with built-in types.\n        String timestampPolicy = config.timestampPolicy;\n        if (timestampPolicy.equals(\"ProcessingTime\")) {\n          builder.setTimestampPolicyFactory(TimestampPolicyFactory.withProcessingTime());\n        } else if (timestampPolicy.equals(\"CreateTime\")) {\n          builder.setTimestampPolicyFactory(TimestampPolicyFactory.withCreateTime(Duration.ZERO));\n        } else if (timestampPolicy.equals(\"LogAppendTime\")) {\n          builder.setTimestampPolicyFactory(TimestampPolicyFactory.withLogAppendTime());\n        } else {\n          throw new IllegalArgumentException(\n              \"timestampPolicy should be one of (ProcessingTime, CreateTime, LogAppendTime)\");\n        }\n\n        if (config.startReadTime != null) {\n          builder.setStartReadTime(Instant.ofEpochMilli(config.startReadTime));\n        }\n\n        if (config.stopReadTime != null) {\n          builder.setStopReadTime(Instant.ofEpochMilli(config.stopReadTime));\n        }\n\n        if (config.dynamicReadPollIntervalSeconds != null) {\n          builder.setDynamicRead(true);\n          builder.setWatchTopicPartitionDuration(\n              Duration.standardSeconds(config.dynamicReadPollIntervalSeconds));\n        } else {\n          builder.setDynamicRead(false);\n        }","sourceCodeStart":888,"sourceCodeEnd":924,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L888-L924","documentation":"KafkaIO's external/cross-language builder (setupExternalBuilder) validates the timestampPolicy string against the three supported values. Any other string cannot be mapped to a TimestampPolicyFactory, so an IllegalArgumentException is thrown.","triggerScenarios":"Passing timestampPolicy in the external KafkaIO configuration (e.g. via SQL/cross-language config) with a value other than exactly 'ProcessingTime', 'CreateTime', or 'LogAppendTime' (case-sensitive).","commonSituations":"Typo such as 'processingTime' or 'CREATETIME'; passing 'CreateTime' variants like 'create_time'; configuring via template/SQL where the value comes from user input.","solutions":["Set timestampPolicy to one of the exact strings: ProcessingTime, CreateTime, or LogAppendTime.","Fix casing — the comparison is case-sensitive.","Omit the field to use the default policy.","Validate user-supplied policy strings before submitting the pipeline."],"exampleFix":"// before\n{\"timestampPolicy\": \"processingTime\"}\n// after\n{\"timestampPolicy\": \"ProcessingTime\"}","handlingStrategy":"validation","validationCode":"if (!Set.of(\"ProcessingTime\",\"CreateTime\",\"LogAppendTime\").contains(policy)) throw new IllegalArgumentException(policy);","typeGuard":null,"tryCatchPattern":"try { build(cfg); } catch (IllegalArgumentException e) { fixPolicy(); }","preventionTips":["Validate enum casing pre-submit"],"tags":["kafka","configuration","enum","beam-io"],"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"}