{"record":{"id":"a4e8170c843a15c5","repo":"apache/beam","slug":"unexpected","errorCode":null,"errorMessage":"{}: Unexpected","messagePattern":"(.+?): Unexpected","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":739,"sourceCode":"   */\n  void finalizeCheckpointMarkAsync(KafkaCheckpointMark checkpointMark) {\n    if (finalizedCheckpointMark.getAndSet(checkpointMark) != null) {\n      checkpointMarkCommitsSkipped.inc();\n    }\n    checkpointMarkCommitsEnqueued.inc();\n  }\n\n  private void nextBatch() throws IOException {\n    curBatch = Collections.emptyIterator();\n\n    ConsumerRecords<byte[], byte[]> records;\n    try {\n      // poll available records, wait (if necessary) up to the specified timeout.\n      records =\n          availableRecordsQueue.poll(recordsDequeuePollTimeout.getMillis(), TimeUnit.MILLISECONDS);\n    } catch (InterruptedException e) {\n      Thread.currentThread().interrupt();\n      LOG.warn(\"{}: Unexpected\", this, e);\n      return;\n    }\n\n    if (records == null) {\n      // Check if the poll thread failed with an exception.\n      if (consumerPollException.get() != null) {\n        throw new IOException(\"Exception while reading from Kafka\", consumerPollException.get());\n      }\n      if (recordsDequeuePollTimeout.isLongerThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MIN)) {\n        recordsDequeuePollTimeout = recordsDequeuePollTimeout.minus(Duration.millis(1));\n        LOG.debug(\"Reducing poll timeout for reader to {}\", recordsDequeuePollTimeout.getMillis());\n      }\n      return;\n    }\n\n    if (recordsDequeuePollTimeout.isShorterThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MAX)) {\n      recordsDequeuePollTimeout = recordsDequeuePollTimeout.plus(Duration.millis(1));\n      LOG.debug(\"Increasing poll timeout for reader to {}\", recordsDequeuePollTimeout.getMillis());","sourceCodeStart":721,"sourceCodeEnd":757,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L721-L757","documentation":"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.","triggerScenarios":"The thread calling KafkaUnboundedReader.advance() is interrupted while blocking on availableRecordsQueue.poll(recordsDequeuePollTimeout), i.e. during runner shutdown or cancellation of the current bundle.","commonSituations":"Pipeline teardown interrupting reader threads; bundle cancellation by the runner; a watchdog interrupting a worker thread perceived as stuck.","solutions":["Ignore if it coincides with pipeline shutdown or bundle cancellation.","If unexpected, audit for stray Thread.interrupt() calls on Beam worker threads.","Check the runner for cancellation/failure events preceding the warning.","Ensure the advance loop retries after early return, as the interrupt flag remains set."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  records = availableRecordsQueue.poll(timeoutMs, TimeUnit.MILLISECONDS);\n} catch (InterruptedException e) {\n  Thread.currentThread().interrupt();\n  LOG.warn(\"Unexpected interrupt while dequeuing records\", e);\n  return Collections.emptyList();\n}","preventionTips":["Restore the interrupt status after catching InterruptedException.","Correlate the warning with runner cancellation events before treating it as a bug.","Keep reader advance loops resilient to early empty returns."],"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"}