{"record":{"id":"86eb580893744c70","repo":"apache/beam","slug":"logappendtimepolicy-policy-is-enabled-in-reader-but-kafka","errorCode":null,"errorMessage":"LogAppendTimePolicy policy is enabled in reader, but Kafka record's timestamp type is LogAppendTime. Most likely it is not enabled on Kafka for the topic '%s'. Actual timestamp type is '%s'.","messagePattern":"LogAppendTimePolicy policy is enabled in reader, but Kafka record's timestamp type is LogAppendTime\\. Most likely it is not enabled on Kafka for the topic '(.+?)'\\. Actual timestamp type is '(.+?)'\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/TimestampPolicyFactory.java","lineNumber":155,"sourceCode":"     * advance the watermark to 'last backlog check time - small delta to account for any internal\n     * buffering in Kafka'. Using 2 seconds for this delta. Should this be user configurable?\n     */\n    private static final Duration IDLE_WATERMARK_DELTA = Duration.standardSeconds(2);\n\n    protected Instant currentWatermark;\n\n    public LogAppendTimePolicy(Optional<Instant> previousWatermark) {\n      currentWatermark = previousWatermark.orElse(BoundedWindow.TIMESTAMP_MIN_VALUE);\n    }\n\n    @Override\n    public Instant getTimestampForRecord(PartitionContext context, KafkaRecord<K, V> record) {\n      if (record.getTimestampType().equals(KafkaTimestampType.LOG_APPEND_TIME)) {\n        currentWatermark = new Instant(record.getTimestamp());\n      } else if (currentWatermark.equals(BoundedWindow.TIMESTAMP_MIN_VALUE)) {\n        // This is the first record and it does not have LOG_APPEND_TIME.\n        // Most likely the topic is not configured correctly.\n        throw new IllegalStateException(\n            String.format(\n                \"LogAppendTimePolicy policy is enabled in reader, but Kafka record's timestamp type \"\n                    + \"is LogAppendTime. Most likely it is not enabled on Kafka for the topic '%s'. \"\n                    + \"Actual timestamp type is '%s'.\",\n                record.getTopic(), record.getTimestampType()));\n      }\n      return currentWatermark;\n    }\n\n    @Override\n    public Instant getWatermark(PartitionContext context) {\n      if (context.getMessageBacklog() == 0) {\n        // The reader is caught up. May need to advance the watermark.\n        Instant idleWatermark = context.getBacklogCheckTime().minus(IDLE_WATERMARK_DELTA);\n        if (idleWatermark.isAfter(currentWatermark)) {\n          currentWatermark = idleWatermark;\n        }\n      } // else, there is backlog (or is unknown). Do not advance the watermark.","sourceCodeStart":137,"sourceCodeEnd":173,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/TimestampPolicyFactory.java#L137-L173","documentation":"Error \"LogAppendTimePolicy policy is enabled in reader, but Kafka record's timestamp type is LogAppendTime. Most likely it is not enabled on Kafka for the topic '%s'. Actual timestamp type is '%s'.\" thrown in apache/beam.","triggerScenarios":"Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/TimestampPolicyFactory.java:155 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["Enable LogAppendTime timestamps on the Kafka topic (log.message.timestamp.type=LogAppendTime) so record timestamps match the policy.","Switch to a different TimestampPolicyFactory (e.g. ProcessingTimePolicy or CreateTimePolicy) that matches the topic's actual timestamp type."],"exampleFix":null,"handlingStrategy":"validation","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":[],"backgroundTag":null,"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"}