{"record":{"id":"b10f55f58522d4c8","repo":"apache/druid","slug":"exception-while-getting-record-from-queue-or-produ","errorCode":null,"errorMessage":"Exception while getting record from queue or producer send, Events would not be emitted anymore.","messagePattern":"Exception while getting record from queue or producer send, Events would not be emitted anymore\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java","lineNumber":224,"sourceCode":"  {\n    sendToKafka(config.getSegmentMetadataTopic(), segmentMetadataQueue, setProducerCallback(segmentMetadataLost));\n  }\n\n  private void sendToKafka(final String topic, MemoryBoundLinkedBlockingQueue<String> recordQueue, Callback callback)\n  {\n    MemoryBoundLinkedBlockingQueue.ObjectContainer<String> objectToSend;\n    try {\n      while (true) {\n        objectToSend = recordQueue.take();\n        producer.send(new ProducerRecord<>(topic, objectToSend.getData()), callback);\n      }\n    }\n    catch (Throwable e) {\n      if (e instanceof InterruptedException && e.getMessage() == null) {\n        log.info(\"Normal exit.\");\n        return;\n      }\n      log.warn(e, \"Exception while getting record from queue or producer send, Events would not be emitted anymore.\");\n    }\n  }\n\n  @Override\n  public void emit(final Event event)\n  {\n    if (event != null) {\n      try {\n        EventMap map = event.toMap();\n        map = addExtraDimensionsToEvent(map);\n\n        String resultJson = jsonMapper.writeValueAsString(map);\n\n        MemoryBoundLinkedBlockingQueue.ObjectContainer<String> objectContainer = new MemoryBoundLinkedBlockingQueue.ObjectContainer<>(\n            resultJson,\n            StringUtils.toUtf8(resultJson).length\n        );\n","sourceCodeStart":206,"sourceCodeEnd":242,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java#L206-L242","documentation":"KafkaEmitter runs a background thread that pulls events from an internal queue and sends them via a KafkaProducer. Any Throwable from queue retrieval or producer.send is caught here and logged with this warning; afterwards the thread exits, meaning no further events will be emitted for the process lifetime until restart.","triggerScenarios":"producer.send fails (broker unreachable, auth failure, record too large, serialization error, timeout), the queue poll throws, or any RuntimeException escapes the loop; an InterruptedException with a null message is treated as normal shutdown exit instead.","commonSituations":"Kafka brokers down or unreachable due to network/DNS issues; wrong bootstrap servers or SASL/SSL credentials in emitter config; messages exceeding max.request.size; Kafka client version mismatch with brokers.","solutions":["Restart the Druid process to restart the emitter thread — once this warning fires, events are permanently dropped for this instance.","Check Kafka connectivity and credentials: verify kafka.producer.config bootstrap.servers, security.protocol, and authentication against the brokers.","Inspect the logged Throwable for the root cause (serialization, size limit, timeout) and fix the producer config accordingly (e.g., raise max.request.size, adjust retries/linger.ms).","Upgrade/patch the emitter to catch per-message errors and continue the loop instead of terminating the thread on the first failure."],"exampleFix":"// before\ncatch (Throwable e) {\n  log.warn(e, \"Exception while getting record from queue or producer send, Events would not be emitted anymore.\");\n}\n// after\ncatch (InterruptedException e) {\n  Thread.currentThread().interrupt();\n  log.info(\"Emitter thread interrupted; exiting.\");\n  return;\n}\ncatch (Exception e) {\n  log.warn(e, \"Failed to send event to Kafka; skipping event and continuing.\");\n}","handlingStrategy":"try-catch","validationCode":"Properties p = new Properties();\np.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);\ntry (Producer<String, byte[]> test = new KafkaProducer<>(p)) {\n  test.partitionsFor(\"__health\"); // throws fast if brokers unreachable/auth bad\n}","typeGuard":"if (e instanceof InterruptedException && e.getMessage() == null) {\n  return; // normal shutdown, not an error\n}","tryCatchPattern":"try {\n  producer.send(record, (md, ex) -> { if (ex != null) log.warn(ex, \"send failed; dropping event\"); });\n} catch (InterruptedException e) {\n  Thread.currentThread().interrupt();\n} catch (Exception e) {\n  log.warn(e, \"Kafka send failed; event dropped\"); // keep thread alive\n}","preventionTips":["Validate Kafka producer config (bootstrap.servers, SASL/SSL) with a smoke producer before deploying.","Set producer retries, acks, and delivery timeouts so transient broker issues self-heal.","Watch kafka-emitter send-failure logs and alert before the thread dies and events are lost.","Size max.request.size/compression to keep emitted events under broker limits."],"tags":["kafka","emitter","event-loss","producer"],"backgroundTag":"network-request-failed","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}