How can I ignore Log4j2 Kafka appender warning messages when broker is not available?

Viewed 661

I have a java appllication which uses log4j2 to log to the console. I wish to send specific log messages from my java application to a kafka server in an aynchronous manner, while still sending other logs to the console. However, when the Kafka server is unavailable, console is being flooded with "Broker may not be available" warning messages, and other logs suposed to go to console seem to wait in queue until the client timeout.

I am using the VM options as instructed in log4j2 documentation for asynchonous logging

-Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector

I have tried setting syncSend to false on the Kafka appender, wraping it with AsyncAppender but nothing seems to work.

This is an example to reproduce the problem

package main;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.Marker;
import org.apache.logging.log4j.MarkerManager;

public class Main {

    public static void main(String[] args) {

        Logger logger = LogManager.getLogger();
        Marker kafkaMarker = MarkerManager.getMarker("KAFKA");

        logger.info("someErrorMessage");

        logger.info(kafkaMarker, "someKafkaMessage");

        for (int i = 0; i < 100; i++) {
            logger.info("someErrorMessage: " + i);
        }
    }
}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>untitled</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <log4j2.version>2.13.3</log4j2.version>
        <java.version>1.8</java.version>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>${log4j2.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${log4j2.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>${log4j2.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-log4j-appender</artifactId>
            <version>2.0.1</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.0.1</version>
        </dependency>
        <dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>3.4.2</version>
        </dependency>
    </dependencies>

</project>

log4j2.xml

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="INFO" strict="true">
    <Properties>
        <Property name="LOG_PATTERN">
            %c.%M %m %ex%n
        </Property>
    </Properties>
    <Appenders>
        <Console name="ConsoleAppender" target="SYSTEM_OUT">
            <Layout type="PatternLayout" pattern="${LOG_PATTERN}"/>
        </Console>
        <Kafka name="KafkaAppender" topic="logProducer" syncSend="false">
            <Property name="bootstrap.servers">localhost:9092</Property>
            <MarkerFilter marker="KAFKA" onMatch="ACCEPT" onMismatch="DENY"/>
            <Layout type="PatternLayout" pattern="${LOG_PATTERN}"/>
        </Kafka>
        <Async name="AsyncKafkaAppender">
            <AppenderRef ref="KafkaAppender"/>
        </Async>
    </Appenders>
    <Loggers>
        <Root level="info">
            <AppenderRef ref="KafkaAppender" />
            <AppenderRef ref="ConsoleAppender" />
        </Root>
    </Loggers>
</Configuration>
1 Answers

There are couple of problems here.

  1. You may decrease max.block.ms parameter to avoid too long blocking. Practically in very fast networks you can go down to 500 or even 100 ms.
  2. You may set very high log level very selective for the kafka class, which sends your errors, so you will not get warnings. However I would not recommend this.

<Logger name="org.apache.kafka" level="ERROR" additivity="false"><AppenderRef ref="Console"/></Logger>

Using the name like org.apache.kafka you can go even deeper with selective logging. Important is to set additivity to false.

Finally I would definitely take care, that Kafka Broker is high available.

Related