apache/hadoop · error · MetricsException

Producer in KafkaSink is null!

Error message

Producer in KafkaSink is null!

What it means

KafkaSink.putMetrics() throws MetricsException('Producer in KafkaSink is null!') whenever it is called with the internal producer field still null. The producer is created only in init(); it is also set to null in close(). So this error means metrics are being pushed either before init() completed (or after it failed at producer creation) or after the sink was closed.

Source

Thrown at hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java:128

    } catch (Exception e) {
      LOG.warn("Error getting Hostname, going to continue");
    }

    System.setProperty("org.apache.kafka.automatic.config.providers", "none");

    try {
      // Create the producer object.
      producer = new KafkaProducer<Integer, byte[]>(props);
    } catch (Exception e) {
      throw new MetricsException("Error creating Producer, " + brokerList, e);
    }
  }

  @Override
  public void putMetrics(MetricsRecord record) {

    if (producer == null) {
      throw new MetricsException("Producer in KafkaSink is null!");
    }

    // Create the json object.
    StringBuilder jsonLines = new StringBuilder();

    long timestamp = record.timestamp();
    Instant instant = Instant.ofEpochMilli(timestamp);
    LocalDateTime ldt = LocalDateTime.ofInstant(instant, zoneId);
    String date = ldt.format(dateFormat);
    String time = ldt.format(timeFormat);

    // Collect datapoints and populate the json object.
    jsonLines.append("{\"hostname\": \"" + hostname);
    jsonLines.append("\", \"timestamp\": " + timestamp);
    jsonLines.append(", \"date\": \"" + date);
    jsonLines.append("\",\"time\": \"" + time);
    jsonLines.append("\",\"name\": \"" + record.name() + "\" ");
    for (MetricsTag tag : record.tags()) {

View on GitHub (pinned to 2add963021)

Solutions

1. Check the logs for a preceding 'Error creating Producer' MetricsException — fixing broker_list/topic usually resolves this follow-on error
2. Guarantee init() completes before the metrics system starts sampling; in tests, inject a producer via KafkaSink.setProducer(mockProducer) before putMetrics
3. Do not reuse a sink after close(); let the metrics framework create a fresh instance

Example fix

// before (unit test)
kafkaSink.putMetrics(record); // MetricsException: producer null

// after
Producer<Integer, byte[]> mock = mock(Producer.class);
kafkaSink.setProducer(mock);
kafkaSink.putMetrics(record);

// before putting metrics (or in a wrapper sink)
if (producer == null) {  // or expose KafkaSink#setProducer in tests
  LOG.debug("Kafka producer not initialized; skipping metrics record");
  return;
}

try {
  sink.putMetrics(record);
} catch (MetricsException e) {
  if (e.getMessage().contains("Producer in KafkaSink is null")) {
    // init failed earlier or sink closed; check for 'Error creating Producer' above
  } else throw e;
}

When it happens

Trigger: The metrics system invoking putMetrics before init() finished (race during sink registration); init() having failed earlier at KafkaProducer creation (see 'Error creating Producer') while the sink object stays registered; putMetrics called again after close() nulled the producer (e.g. during shutdown, metrics flush after close, or a re-registered sink instance).

Common situations: Startup races in metrics sinks on busy daemons; an earlier init failure logged but unnoticed, then every putMetrics throws this; metrics emitted during daemon shutdown after the sink was closed; unit tests calling putMetrics without init (use setProducer() to inject a mock).

Related errors

• {} already exists!
• Metrics source {} already exists!
• Kafka topic can not be null
• Error creating Producer, {}
• Error sending data

AI-assisted analysis of apache/hadoop@2add963021 (2026-08-22). Data as JSON: /api/errors/9d0253ae06466c28. Report an issue: GitHub.