{"record":{"id":"ec69f2e99bea2b1a","repo":"apache/hadoop","slug":"error-creating-producer","errorCode":null,"errorMessage":"Error creating Producer, {}","messagePattern":"Error creating Producer, (.+?)","errorType":"exception","errorClass":"MetricsException","httpStatus":null,"severity":"error","filePath":"hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java","lineNumber":120,"sourceCode":"    props.put(\"value.serializer\",\n        \"org.apache.kafka.common.serialization.ByteArraySerializer\");\n    props.put(\"request.required.acks\", \"0\");\n\n    // Set the hostname once and use it in every message.\n    hostname = \"null\";\n    try {\n      hostname = InetAddress.getLocalHost().getHostName();\n    } 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);","sourceCodeStart":102,"sourceCodeEnd":138,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java#L102-L138","documentation":"During init(), KafkaSink constructs a KafkaProducer with the configured properties (bootstrap.servers from broker_list, byte-array serializers). Any exception from the constructor — unreachable/invalid broker list, missing or incompatible kafka-clients classes, bad producer config — is rethrown as MetricsException('Error creating Producer, <brokerList>'). This is a startup-time failure: the sink cannot function without a producer.","triggerScenarios":"broker_list empty, malformed (e.g. 'kafka1:9092;' instead of comma-separated), or pointing at hosts that fail DNS resolution (KafkaProducer constructor can fail on bad bootstrap config); kafka-clients.jar missing from the classpath or a version conflicting with the one hadoop-kafka was built against; ConfigException from invalid producer properties.","commonSituations":"Deploying the metrics sink without adding kafka-clients.jar to the classpath; broker hostnames not resolvable from the Hadoop node; firewall/proxy setups; mixing Kafka client versions (e.g. old producer configs like request.required.acks against newer clients); typo in broker_list format.","solutions":["Validate broker_list is a comma-separated host:port list (e.g. kafka1:9092,kafka2:9092) and reachable from the Hadoop node (nc -vz kafka1 9092)","Ensure a compatible kafka-clients jar is on the classpath of every daemon loading the sink","Inspect the cause chain (MetricsException wraps the original) — ConfigException points at the exact bad property","Temporarily set *.sink.kafka.* off or point metrics to a file sink until Kafka connectivity is fixed"],"exampleFix":"# before\nresourcemanager.sink.kafka.broker_list=kafka1:9092;\n\n# after\nresourcemanager.sink.kafka.broker_list=kafka1:9092,kafka2:9092","handlingStrategy":"try-catch","validationCode":"// Pre-flight broker reachability before enabling the sink\nfor (String hp : brokerList.split(\",\")) {\n  String[] parts = hp.split(\":\");\n  try (Socket s = new Socket()) {\n    s.connect(new InetSocketAddress(parts[0], Integer.parseInt(parts[1])), 2000);\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  sink.init(conf);\n} catch (MetricsException e) {\n  // cause chain holds the KafkaProducer failure: fix classpath or broker_list\n  LOG.error(\"Kafka sink init failed\", e.getCause());\n}","preventionTips":["Ship a compatible kafka-clients jar with the sink on every daemon's classpath","Verify DNS + network reachability of every broker from Hadoop nodes at deploy time","Use comma-separated host:port pairs in broker_list"],"tags":["hadoop","metrics2","kafka","producer","initialization"],"backgroundTag":"client-initialization-failed","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-22T20:17:22.307Z"}