apache/beam · info

: Unexpected

Error message

{}: Unexpected

What it means

Warning logged in KafkaUnboundedReader when the dequeue of available records from the internal queue is interrupted unexpectedly. The thread re-asserts the interrupt flag and returns early from advance(), yielding no records for this call.

Solutions

1. Ignore if it coincides with pipeline shutdown or bundle cancellation.
2. If unexpected, audit for stray Thread.interrupt() calls on Beam worker threads.
3. Check the runner for cancellation/failure events preceding the warning.
4. Ensure the advance loop retries after early return, as the interrupt flag remains set.

try {
  records = availableRecordsQueue.poll(timeoutMs, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
  Thread.currentThread().interrupt();
  LOG.warn("Unexpected interrupt while dequeuing records", e);
  return Collections.emptyList();
}

When it happens

Trigger: The thread calling KafkaUnboundedReader.advance() is interrupted while blocking on availableRecordsQueue.poll(recordsDequeuePollTimeout), i.e. during runner shutdown or cancellation of the current bundle.

Common situations: Pipeline teardown interrupting reader threads; bundle cancellation by the runner; a watchdog interrupting a worker thread perceived as stuck.

Related errors

• : consumer thread is interrupted
• : closing producer after unrecoverable error. The work…
• consumerPollingTimeout should be > 0.
• Couldn't infer Coder from
• Error while parsing the element

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:739

   */
  void finalizeCheckpointMarkAsync(KafkaCheckpointMark checkpointMark) {
    if (finalizedCheckpointMark.getAndSet(checkpointMark) != null) {
      checkpointMarkCommitsSkipped.inc();
    }
    checkpointMarkCommitsEnqueued.inc();
  }

  private void nextBatch() throws IOException {
    curBatch = Collections.emptyIterator();

    ConsumerRecords<byte[], byte[]> records;
    try {
      // poll available records, wait (if necessary) up to the specified timeout.
      records =
          availableRecordsQueue.poll(recordsDequeuePollTimeout.getMillis(), TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.warn("{}: Unexpected", this, e);
      return;
    }

    if (records == null) {
      // Check if the poll thread failed with an exception.
      if (consumerPollException.get() != null) {
        throw new IOException("Exception while reading from Kafka", consumerPollException.get());
      }
      if (recordsDequeuePollTimeout.isLongerThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MIN)) {
        recordsDequeuePollTimeout = recordsDequeuePollTimeout.minus(Duration.millis(1));
        LOG.debug("Reducing poll timeout for reader to {}", recordsDequeuePollTimeout.getMillis());
      }
      return;
    }

    if (recordsDequeuePollTimeout.isShorterThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MAX)) {
      recordsDequeuePollTimeout = recordsDequeuePollTimeout.plus(Duration.millis(1));
      LOG.debug("Increasing poll timeout for reader to {}", recordsDequeuePollTimeout.getMillis());

View on GitHub (pinned to 12126d8942)