zhisheng17/flink-learning · error · ConfigException

Source must be specified by the Kafka log4j appender

Error message

Source must be specified by the Kafka log4j appender

What it means

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.

Source

Thrown at flink-learning-extends/FlinkLogKafkaAppender/Log4jKafkaAppender/src/main/java/com/zhisheng/log/appender/KafkaLog4jAppender.java:98

            }
        } else {
            log.error("log.file Property ({}) doesn't contains yarn application id or container id", logFile);
        }

        taskId = envProperties.getProperty("taskId", null);
        taskName = envProperties.getProperty("taskName", null);

        Properties props = new Properties();
        if (this.bootstrapServers != null) {
            props.setProperty("bootstrap.servers", this.bootstrapServers);
        } else {
            throw new ConfigException("The bootstrap servers property must be specified");
        }
        if (this.topic == null) {
            throw new ConfigException("Topic must be specified by the Kafka log4j appender");
        }
        if (this.source == null) {
            throw new ConfigException("Source must be specified by the Kafka log4j appender");
        }

        String clientIdPrefix = taskId != null ? taskId : appId;

        if (clientIdPrefix != null) {
            props.setProperty("client.id", clientIdPrefix + "_log");
        }

        if (this.acks != null) {
            props.setProperty("acks", this.acks);
        } else {
            props.setProperty("acks", "0");
        }

        if (this.retries != null) {
            props.setProperty("retries", this.retries);
        } else {
            props.setProperty("retries", "0");

View on GitHub (pinned to d731cee761)

Solutions

  1. Add log4j.appender.<name>.Source=your-app-name to log4j.properties.
  2. Set it to a meaningful value (Flink job name / application id) for traceability.
  3. Align the team's log4j.properties template so Source is always present.
  4. Verify initialization logs no ConfigException after the change.

Example fix

// before
log4j.appender.kafka.Topic=app-logs
// after
log4j.appender.kafka.Topic=app-logs
log4j.appender.kafka.Source=my-flink-job
Defensive patterns

Strategy: validation

Validate before calling

java.util.Properties p = new java.util.Properties();
p.load(new FileInputStream("log4j.properties"));
if (p.getProperty("log4j.appender.kafka.Source") == null)
    throw new IllegalStateException("Source missing for KafkaLog4jAppender");

Try / catch

try {
    PropertyConfigurator.configure("log4j.properties");
} catch (Throwable t) {
    System.err.println("Kafka appender missing Source: " + t.getMessage());
}

Prevention

When it happens

Trigger: log4j.properties appender config missing the Source option while BootstrapServers and Topic are present.

Common situations: 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.

Understand the failure class

Background: "X is required", "must be set", "cannot be empty": the missing-required-config error family, from Vertex AI project/location to WeChat keys — this error's family across 18 libraries.

Related errors

AI-assisted analysis of zhisheng17/flink-learning@d731cee761 (2026-09-06). Data as JSON: /api/errors/8de46bf2872361fb. Report an issue: GitHub.