Exception while getting record from queue or producer send…

Error message

Exception while getting record from queue or producer send, Events would not be emitted anymore.

What it means

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.

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.

Example fix

// before
catch (Throwable e) {
  log.warn(e, "Exception while getting record from queue or producer send, Events would not be emitted anymore.");
}
// after
catch (InterruptedException e) {
  Thread.currentThread().interrupt();
  log.info("Emitter thread interrupted; exiting.");
  return;
}
catch (Exception e) {
  log.warn(e, "Failed to send event to Kafka; skipping event and continuing.");
}

Defensive patterns

Strategy: try-catch

Validate before calling

Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
try (Producer<String, byte[]> test = new KafkaProducer<>(p)) {
  test.partitionsFor("__health"); // throws fast if brokers unreachable/auth bad
}

Type guard

if (e instanceof InterruptedException && e.getMessage() == null) {
  return; // normal shutdown, not an error
}

Try / catch

try {
  producer.send(record, (md, ex) -> { if (ex != null) log.warn(ex, "send failed; dropping event"); });
} catch (InterruptedException e) {
  Thread.currentThread().interrupt();
} catch (Exception e) {
  log.warn(e, "Kafka send failed; event dropped"); // keep thread alive
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/b10f55f58522d4c8. Report an issue: GitHub.

Appendix: source

Thrown at extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java:224

  {
    sendToKafka(config.getSegmentMetadataTopic(), segmentMetadataQueue, setProducerCallback(segmentMetadataLost));
  }

  private void sendToKafka(final String topic, MemoryBoundLinkedBlockingQueue<String> recordQueue, Callback callback)
  {
    MemoryBoundLinkedBlockingQueue.ObjectContainer<String> objectToSend;
    try {
      while (true) {
        objectToSend = recordQueue.take();
        producer.send(new ProducerRecord<>(topic, objectToSend.getData()), callback);
      }
    }
    catch (Throwable e) {
      if (e instanceof InterruptedException && e.getMessage() == null) {
        log.info("Normal exit.");
        return;
      }
      log.warn(e, "Exception while getting record from queue or producer send, Events would not be emitted anymore.");
    }
  }

  @Override
  public void emit(final Event event)
  {
    if (event != null) {
      try {
        EventMap map = event.toMap();
        map = addExtraDimensionsToEvent(map);

        String resultJson = jsonMapper.writeValueAsString(map);

        MemoryBoundLinkedBlockingQueue.ObjectContainer<String> objectContainer = new MemoryBoundLinkedBlockingQueue.ObjectContainer<>(
            resultJson,
            StringUtils.toUtf8(resultJson).length
        );

View on GitHub (pinned to 9b90983fd2)