apache/beam · error · IllegalStateException

LogAppendTimePolicy policy is enabled in reader, but Kafka…

Error message

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'.

What it means

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.

Solutions

1. Enable LogAppendTime timestamps on the Kafka topic (log.message.timestamp.type=LogAppendTime) so record timestamps match the policy.
2. Switch to a different TimestampPolicyFactory (e.g. ProcessingTimePolicy or CreateTimePolicy) that matches the topic's actual timestamp type.

When it happens

Trigger: 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.

Common situations: See trigger scenarios.

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/TimestampPolicyFactory.java:155

     * advance the watermark to 'last backlog check time - small delta to account for any internal
     * buffering in Kafka'. Using 2 seconds for this delta. Should this be user configurable?
     */
    private static final Duration IDLE_WATERMARK_DELTA = Duration.standardSeconds(2);

    protected Instant currentWatermark;

    public LogAppendTimePolicy(Optional<Instant> previousWatermark) {
      currentWatermark = previousWatermark.orElse(BoundedWindow.TIMESTAMP_MIN_VALUE);
    }

    @Override
    public Instant getTimestampForRecord(PartitionContext context, KafkaRecord<K, V> record) {
      if (record.getTimestampType().equals(KafkaTimestampType.LOG_APPEND_TIME)) {
        currentWatermark = new Instant(record.getTimestamp());
      } else if (currentWatermark.equals(BoundedWindow.TIMESTAMP_MIN_VALUE)) {
        // This is the first record and it does not have LOG_APPEND_TIME.
        // Most likely the topic is not configured correctly.
        throw new IllegalStateException(
            String.format(
                "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'.",
                record.getTopic(), record.getTimestampType()));
      }
      return currentWatermark;
    }

    @Override
    public Instant getWatermark(PartitionContext context) {
      if (context.getMessageBacklog() == 0) {
        // The reader is caught up. May need to advance the watermark.
        Instant idleWatermark = context.getBacklogCheckTime().minus(IDLE_WATERMARK_DELTA);
        if (idleWatermark.isAfter(currentWatermark)) {
          currentWatermark = idleWatermark;
        }
      } // else, there is backlog (or is unknown). Do not advance the watermark.

View on GitHub (pinned to 12126d8942)