{"record":{"id":"8de46bf2872361fb","repo":"zhisheng17/flink-learning","slug":"source-must-be-specified-by-the-kafka-log4j-append","errorCode":null,"errorMessage":"Source must be specified by the Kafka log4j appender","messagePattern":"Source must be specified by the Kafka log4j appender","errorType":"validation","errorClass":"ConfigException","httpStatus":null,"severity":"error","filePath":"flink-learning-extends/FlinkLogKafkaAppender/Log4jKafkaAppender/src/main/java/com/zhisheng/log/appender/KafkaLog4jAppender.java","lineNumber":98,"sourceCode":"            }\n        } else {\n            log.error(\"log.file Property ({}) doesn't contains yarn application id or container id\", logFile);\n        }\n\n        taskId = envProperties.getProperty(\"taskId\", null);\n        taskName = envProperties.getProperty(\"taskName\", null);\n\n        Properties props = new Properties();\n        if (this.bootstrapServers != null) {\n            props.setProperty(\"bootstrap.servers\", this.bootstrapServers);\n        } else {\n            throw new ConfigException(\"The bootstrap servers property must be specified\");\n        }\n        if (this.topic == null) {\n            throw new ConfigException(\"Topic must be specified by the Kafka log4j appender\");\n        }\n        if (this.source == null) {\n            throw new ConfigException(\"Source must be specified by the Kafka log4j appender\");\n        }\n\n        String clientIdPrefix = taskId != null ? taskId : appId;\n\n        if (clientIdPrefix != null) {\n            props.setProperty(\"client.id\", clientIdPrefix + \"_log\");\n        }\n\n        if (this.acks != null) {\n            props.setProperty(\"acks\", this.acks);\n        } else {\n            props.setProperty(\"acks\", \"0\");\n        }\n\n        if (this.retries != null) {\n            props.setProperty(\"retries\", this.retries);\n        } else {\n            props.setProperty(\"retries\", \"0\");","sourceCodeStart":80,"sourceCodeEnd":116,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-extends/FlinkLogKafkaAppender/Log4jKafkaAppender/src/main/java/com/zhisheng/log/appender/KafkaLog4jAppender.java#L80-L116","documentation":"Thrown as an org.apache.kafka.common.config.ConfigException from KafkaLog4jAppender.activateOptions() when the log4j configuration for this appender lacks the `source` property (this.source == null). activateOptions is invoked once during appender initialization from log4j.properties/xml, so an unset Source value stops the appender from being activated and no log events are forwarded to Kafka. This is a generic null check of a user-supplied configuration key — the at-fault input is the appender's `source` config property missing from the log4j configuration file.","triggerScenarios":"log4j.properties appender config missing the Source option while BootstrapServers and Topic are present.","commonSituations":"Upgrading from a plain Kafka appender that didn't need Source; team config template lacks Source; Source spelled 'source' vs expected casing in properties keys.","solutions":["Add log4j.appender.<name>.Source=your-app-name to log4j.properties.","Set it to a meaningful value (Flink job name / application id) for traceability.","Align the team's log4j.properties template so Source is always present.","Verify initialization logs no ConfigException after the change."],"exampleFix":"// before\nlog4j.appender.kafka.Topic=app-logs\n// after\nlog4j.appender.kafka.Topic=app-logs\nlog4j.appender.kafka.Source=my-flink-job","handlingStrategy":"validation","validationCode":"java.util.Properties p = new java.util.Properties();\np.load(new FileInputStream(\"log4j.properties\"));\nif (p.getProperty(\"log4j.appender.kafka.Source\") == null)\n    throw new IllegalStateException(\"Source missing for KafkaLog4jAppender\");","typeGuard":null,"tryCatchPattern":"try {\n    PropertyConfigurator.configure(\"log4j.properties\");\n} catch (Throwable t) {\n    System.err.println(\"Kafka appender missing Source: \" + t.getMessage());\n}","preventionTips":["Always set Source (application/job identifier) in the appender config","Template the properties file so Source defaults to the deploy name","Document required options: BootstrapServers, Topic, Source","Test appender init in staging before production rollout"],"tags":["kafka","log4j","config"],"backgroundTag":"missing-required-config","analyzedSha":"d731cee7618021be56d132cc925102ffff8d75e6","analyzedAt":"2026-09-06T05:35:08.496Z","contentChangedAt":"2026-09-06T05:35:08.496Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}