{"record":{"id":"25f22b503c7dcff9","repo":"apache/beam","slug":"consumer-thread-is-interrupted","errorCode":null,"errorMessage":"{}: consumer thread is interrupted","messagePattern":"(.+?): consumer thread is interrupted","errorType":"console","errorClass":null,"httpStatus":null,"severity":"info","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java","lineNumber":612,"sourceCode":"      while (!closed.get()) {\n        try {\n          if (records.isEmpty()) {\n            stopwatch.start();\n            records = consumer.poll(KAFKA_POLL_TIMEOUT.getMillis());\n            stopwatch.stop();\n            for (String kafkaTopic : kafkaTopics) {\n              kafkaResults.updateSuccessfulRpcMetrics(\n                  kafkaTopic,\n                  java.time.Duration.ofMillis(stopwatch.elapsed(TimeUnit.MILLISECONDS)));\n            }\n          } else if (availableRecordsQueue.offer(\n              records, RECORDS_ENQUEUE_POLL_TIMEOUT.getMillis(), TimeUnit.MILLISECONDS)) {\n            records = ConsumerRecords.empty();\n          }\n\n          commitCheckpointMark();\n        } catch (InterruptedException e) {\n          LOG.warn(\"{}: consumer thread is interrupted\", this, e); // not expected\n          break;\n        } catch (WakeupException e) {\n          break;\n        }\n      }\n      LOG.info(\"{}: Returning from consumer pool loop\", this);\n    } catch (Exception e) { // mostly an unrecoverable KafkaException.\n      LOG.error(\"{}: Exception while reading from Kafka\", this, e);\n      consumerPollException.set(e);\n      throw e;\n    }\n  }\n\n  @SuppressWarnings(\"Slf4jFormatShouldBeConst\")\n  private void commitCheckpointMark() {\n    KafkaCheckpointMark checkpointMark = finalizedCheckpointMark.getAndSet(null);\n    if (checkpointMark == null) {\n      return;","sourceCodeStart":594,"sourceCodeEnd":630,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L594-L630","documentation":"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.","triggerScenarios":"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.","commonSituations":"Beam runner draining or tearing down the worker; a bundle cancelled mid-run; watchdog/watchdog thread interrupting a stuck consumer during pipeline termination.","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."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  // consumer loop work\n} catch (InterruptedException e) {\n  Thread.currentThread().interrupt(); // preserve interrupt status\n  LOG.warn(\"Consumer thread interrupted; shutting down\", e);\n  break;\n}","preventionTips":["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."],"tags":["kafka","thread-interruption","consumer","beam-io"],"backgroundTag":"thread-interrupted","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"}