zhisheng17/flink-learning · error · ConfigException

Topic must be specified by the Kafka log4j appender

Error message

Topic 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 `topic` property (this.topic == null). activateOptions runs once at appender initialization when log4j parses log4j.properties/log4j.xml, so a missing Topic line in the appender config aborts appender startup and the appender cannot deliver any log events to Kafka. The guard is a plain null validation of a user-supplied config key — the at-fault input is the appender's `topic` config property being absent (or empty/unset) in the log4j configuration file.

Source

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

                containerType = "taskmanager";
            } else {
                containerType = "others";
            }
        } 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) {

View on GitHub (pinned to d731cee761)

Solutions

  1. Add log4j.appender.<name>.Topic=your-topic to log4j.properties.
  2. Make sure the prefix matches the appender name you actually registered.
  3. Create the topic on the broker if it does not exist (or rely on auto-create).
  4. Restart the app so activateOptions() re-runs with valid config.

Example fix

// before
log4j.appender.kafka.BootstrapServers=broker:9092
// after
log4j.appender.kafka.BootstrapServers=broker:9092
log4j.appender.kafka.Topic=app-logs
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.Topic") == null)
    throw new IllegalStateException("Topic missing for KafkaLog4jAppender");

Try / catch

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

Prevention

When it happens

Trigger: Appender configured in log4j.properties with BootstrapServers but no Topic option, at logger initialization time.

Common situations: Forgotten Topic line; copy-pasted appender config stripped of the Topic; topic set on the wrong appender prefix (e.g. wrong appender name in the key).

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/05e4edfb54544672. Report an issue: GitHub.