What causes "{}: consumer thread is interrupted" in apache/beam?

Treat as an expected consequence of pipeline shutdown; no action needed if the pipeline is stopping.

: consumer thread is interrupted

Error message

{}: consumer thread is interrupted

What it means

Warning logged in KafkaUnboundedReader's consumer pool loop when the consumer thread's poll loop is interrupted while enqueuing records or committing a checkpoint. The code comments this as unexpected; the loop breaks and the consumer thread exits.

Solutions

Treat as an expected consequence of pipeline shutdown; no action needed if the pipeline is stopping.
If it appears during normal (non-shutdown) operation, look for code calling Thread.interrupt() on worker threads and remove it.
Check runner logs for bundle cancellation/failure just before the interrupt to find the root cause.
If commits were missed at shutdown, verify checkpointing/restart behavior for the source.

Defensive patterns

Strategy: try-catch

Try / catch

try {
  // consumer loop work
} catch (InterruptedException e) {
  Thread.currentThread().interrupt(); // preserve interrupt status
  LOG.warn("Consumer thread interrupted; shutting down", e);
  break;
}

Prevention

Expect interrupts during drain/teardown; log at info/warn and exit cleanly.
Never swallow InterruptedException without restoring the interrupt flag.
Investigate stray interrupt sources only if the warning occurs outside shutdown.

When it happens

Trigger: The thread running the KafkaUnboundedReader consumer loop is interrupted via Thread.interrupt() while waiting to enqueue records (poll with RECORDS_ENQUEUE_POLL_TIMEOUT) or while committing a checkpoint mark — typically during pipeline shutdown or runner-driven cancellation.

Common situations: Beam runner draining or tearing down the worker; a bundle cancelled mid-run; watchdog/watchdog thread interrupting a stuck consumer during pipeline termination.

Related errors

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

Appendix: source

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

      while (!closed.get()) {
        try {
          if (records.isEmpty()) {
            stopwatch.start();
            records = consumer.poll(KAFKA_POLL_TIMEOUT.getMillis());
            stopwatch.stop();
            for (String kafkaTopic : kafkaTopics) {
              kafkaResults.updateSuccessfulRpcMetrics(
                  kafkaTopic,
                  java.time.Duration.ofMillis(stopwatch.elapsed(TimeUnit.MILLISECONDS)));
            }
          } else if (availableRecordsQueue.offer(
              records, RECORDS_ENQUEUE_POLL_TIMEOUT.getMillis(), TimeUnit.MILLISECONDS)) {
            records = ConsumerRecords.empty();
          }

          commitCheckpointMark();
        } catch (InterruptedException e) {
          LOG.warn("{}: consumer thread is interrupted", this, e); // not expected
          break;
        } catch (WakeupException e) {
          break;
        }
      }
      LOG.info("{}: Returning from consumer pool loop", this);
    } catch (Exception e) { // mostly an unrecoverable KafkaException.
      LOG.error("{}: Exception while reading from Kafka", this, e);
      consumerPollException.set(e);
      throw e;
    }
  }

  @SuppressWarnings("Slf4jFormatShouldBeConst")
  private void commitCheckpointMark() {
    KafkaCheckpointMark checkpointMark = finalizedCheckpointMark.getAndSet(null);
    if (checkpointMark == null) {
      return;

View on GitHub (pinned to 12126d8942)