{"record":{"id":"f7ce04cf2735b532","repo":"apache/hadoop","slug":"kafka-topic-can-not-be-null","errorCode":null,"errorMessage":"Kafka topic can not be null","messagePattern":"Kafka topic can not be 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":96,"sourceCode":"  public void init(SubsetConfiguration conf) {\n    // Get Kafka broker configuration.\n    Properties props = new Properties();\n    brokerList = conf.getString(BROKER_LIST);\n    if (LOG.isDebugEnabled()) {\n      LOG.debug(\"Broker list \" + brokerList);\n    }\n    props.put(\"bootstrap.servers\", brokerList);\n    if (LOG.isDebugEnabled()) {\n      LOG.debug(\"Kafka brokers: \" + brokerList);\n    }\n\n    // Get Kafka topic configuration.\n    topic = conf.getString(TOPIC);\n    if (LOG.isDebugEnabled()) {\n      LOG.debug(\"Kafka topic \" + topic);\n    }\n    if (Strings.isNullOrEmpty(topic)) {\n      throw new MetricsException(\"Kafka topic can not be null\");\n    }\n\n    // Set the rest of Kafka configuration.\n    props.put(\"key.serializer\",\n        \"org.apache.kafka.common.serialization.ByteArraySerializer\");\n    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\");","sourceCodeStart":78,"sourceCodeEnd":114,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java#L78-L114","documentation":"KafkaSink.init() reads the 'topic' property from the metrics2 sink configuration (hadoop-metrics2.properties, e.g. *.sink.kafka.topic). If it is null or empty (Strings.isNullOrEmpty), init throws MetricsException('Kafka topic can not be null'). The sink publishes every metrics record to this single Kafka topic, so the property is mandatory.","triggerScenarios":"Configuring a kafka sink block in hadoop-metrics2.properties that sets broker_list but omits topic (or sets it to an empty string); misspelling the key (topics, kafka.topic) so conf.getString(TOPIC) returns null; enabling the sink via *.sink.kafka.class without any properties.","commonSituations":"First-time setup of the hadoop-kafka metrics sink; copying an example config that comments out the topic line; property-name drift between Hadoop versions or documentation typos; trailing whitespace-only values are not caught by IsNullOrEmpty and will fail later at send time instead.","solutions":["Add the topic key to the sink block: *.sink.kafka.topic=<your-topic> alongside *.sink.kafka.broker_list","Check exact spelling: the keys are 'broker_list' and 'topic' under the sink prefix (e.g. resourcemanager.sink.kafka.topic)","Verify the topic exists on the Kafka cluster (or auto-create is enabled) so later sends do not fail"],"exampleFix":"# before (hadoop-metrics2.properties)\nresourcemanager.sink.kafka.class=org.apache.hadoop.metrics2.sink.KafkaSink\nresourcemanager.sink.kafka.broker_list=kafka1:9092,kafka2:9092\n\n# after\nresourcemanager.sink.kafka.class=org.apache.hadoop.metrics2.sink.KafkaSink\nresourcemanager.sink.kafka.broker_list=kafka1:9092,kafka2:9092\nresourcemanager.sink.kafka.topic=hadoop-metrics","handlingStrategy":"validation","validationCode":"// Validate hadoop-metrics2.properties before deploy\nProperties p = load(\"hadoop-metrics2.properties\");\nString topic = p.getProperty(\"*.sink.kafka.topic\");\nif (topic == null || topic.trim().isEmpty()) {\n  throw new IllegalStateException(\"KafkaSink requires *.sink.kafka.topic\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Template every kafka sink block with both broker_list and topic keys filled in","Spell keys exactly: 'broker_list' and 'topic' under the sink prefix","Config-lint hadoop-metrics2.properties in deployment scripts"],"tags":["hadoop","metrics2","kafka","sink","configuration"],"backgroundTag":"missing-config-key","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-23T01:17:44.959Z"}