{"record":{"id":"9d0253ae06466c28","repo":"apache/hadoop","slug":"producer-in-kafkasink-is-null","errorCode":null,"errorMessage":"Producer in KafkaSink is null!","messagePattern":"Producer in KafkaSink is null!","errorType":"exception","errorClass":"MetricsException","httpStatus":null,"severity":"error","filePath":"hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java","lineNumber":128,"sourceCode":"    } catch (Exception e) {\n      LOG.warn(\"Error getting Hostname, going to continue\");\n    }\n\n    System.setProperty(\"org.apache.kafka.automatic.config.providers\", \"none\");\n\n    try {\n      // Create the producer object.\n      producer = new KafkaProducer<Integer, byte[]>(props);\n    } catch (Exception e) {\n      throw new MetricsException(\"Error creating Producer, \" + brokerList, e);\n    }\n  }\n\n  @Override\n  public void putMetrics(MetricsRecord record) {\n\n    if (producer == null) {\n      throw new MetricsException(\"Producer in KafkaSink is null!\");\n    }\n\n    // Create the json object.\n    StringBuilder jsonLines = new StringBuilder();\n\n    long timestamp = record.timestamp();\n    Instant instant = Instant.ofEpochMilli(timestamp);\n    LocalDateTime ldt = LocalDateTime.ofInstant(instant, zoneId);\n    String date = ldt.format(dateFormat);\n    String time = ldt.format(timeFormat);\n\n    // Collect datapoints and populate the json object.\n    jsonLines.append(\"{\\\"hostname\\\": \\\"\" + hostname);\n    jsonLines.append(\"\\\", \\\"timestamp\\\": \" + timestamp);\n    jsonLines.append(\", \\\"date\\\": \\\"\" + date);\n    jsonLines.append(\"\\\",\\\"time\\\": \\\"\" + time);\n    jsonLines.append(\"\\\",\\\"name\\\": \\\"\" + record.name() + \"\\\" \");\n    for (MetricsTag tag : record.tags()) {","sourceCodeStart":110,"sourceCodeEnd":146,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java#L110-L146","documentation":"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.","triggerScenarios":"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).","commonSituations":"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).","solutions":["Check the logs for a preceding 'Error creating Producer' MetricsException — fixing broker_list/topic usually resolves this follow-on error","Guarantee init() completes before the metrics system starts sampling; in tests, inject a producer via KafkaSink.setProducer(mockProducer) before putMetrics","Do not reuse a sink after close(); let the metrics framework create a fresh instance"],"exampleFix":"// before (unit test)\nkafkaSink.putMetrics(record); // MetricsException: producer null\n\n// after\nProducer<Integer, byte[]> mock = mock(Producer.class);\nkafkaSink.setProducer(mock);\nkafkaSink.putMetrics(record);","handlingStrategy":"validation","validationCode":"// before putting metrics (or in a wrapper sink)\nif (producer == null) {  // or expose KafkaSink#setProducer in tests\n  LOG.debug(\"Kafka producer not initialized; skipping metrics record\");\n  return;\n}","typeGuard":null,"tryCatchPattern":"try {\n  sink.putMetrics(record);\n} catch (MetricsException e) {\n  if (e.getMessage().contains(\"Producer in KafkaSink is null\")) {\n    // init failed earlier or sink closed; check for 'Error creating Producer' above\n  } else throw e;\n}","preventionTips":["In tests, inject a mock via KafkaSink.setProducer() before putMetrics","Never reuse a sink after close(); let the metrics framework rebuild it","Treat this as a symptom — grep logs for the root init failure"],"tags":["hadoop","metrics2","kafka","lifecycle","null-check"],"backgroundTag":"uninitialized-client","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-23T01:17:44.959Z"}