Transmitting Log Messages to a Kafka Topic Using Kafka Log4j2 Appender
In this article, learn more about Log4j, a popular logging package for Java development that offers a simple, comprehensive, and configurable logging system.
Context
It could be a basic "Hello, World!" application or a complex banking solution like Stripe, but developing an application is a fascinating process. This process typically includes extensive testing and quality assurance to ensure that not only the requirements are met but also that the application is reliable enough for users to consume.
While this procedure is thorough, faults and errors do occur over time as users interact with the program, and the application must be quickly troubleshot to remedy these bugs or mistakes.
Logs are generated by logging a collection of events that occur with the program, and they provide a way to identify which element of the application process is not operating as planned. Log4j is a popular logging package for Java development that offers a simple, comprehensive, and configurable logging system.
Kafka is the most popular open-source stream-processing program for gathering, processing, storing, and analyzing data at scale. It can handle hundreds of messages per second and is noted for its exceptional performance, low latency, fault tolerance, and high throughput.
Prerequisites
Java 8 installed
Docker installed
Basic understanding of Log4j
Basic understanding of Kafka
Steps
- Create a topic for Log4j to subscribe to and produce log messages by creating a docker-compose file with the snippet below. Then, run docker-compose up -d to spin up Kafka and ZooKeeper containers.
Dockerfile
version: '2.1'
services:
zookeeper:
image: confluentinc/cp-zookeeper:6.1.1
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
# reachable on 9092 from the host and on 29092 from inside docker compose
kafka:
image: confluentinc/cp-kafka:6.1.1
depends_on:
- zookeeper
ports:
- '9092:9092'
expose:
- '29092'
environment:
KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: '1'
KAFKA_MIN_INSYNC_REPLICAS: '1'
init-kafka:
image: confluentinc/cp-kafka:6.1.1
depends_on:
- kafka
entrypoint: [ '/bin/sh', '-c' ]
command: |
"
# blocks until kafka is reachable
kafka-topics --bootstrap-server kafka:29092 --list
echo -e 'Creating kafka topics'
kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic my-topic-1 --replication-factor 1 --partitions 1
echo -e 'Successfully created the following topics:'
kafka-topics --bootstrap-server kafka:29092 --list
"- Create a Spring Boot application with Maven, add kafka-log4j-appender,
kafkaas a dependency, and exclude Spring Boot default logging properties. The POM file should be like this:
XML
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-log4j-appender</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>1.0.1</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>- Create a
HelloWorldclass that simply logs“Hello World sent to Kafka topic at”at a particular interval using the Log4j logger and invoke thecreateLogs()method in the main class. This will serve as a producer who will publish the log messages to the Kafka topic created with the docker-compose file.
Java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
public class HelloWorld {
private static final Logger logger = LoggerFactory.getLogger(HelloWorld.class);
public static void createLogs() throws InterruptedException {
int k = 0;
boolean b = true;
while (b) {
logger.info("Hello World sent to Kafka topic at " +
DateTimeFormatter.ofPattern("HH:mm:ss").format(LocalDateTime.now()));
k++;
b = k < Integer.MAX_VALUE;
Thread.sleep(1000);
}
}
}
Java
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Log4jAppenderApplication {
public static void main(String[] args) throws InterruptedException {
SpringApplication.run(Log4jAppenderApplication.class, args);
HelloWorld helloWorld = new HelloWorld();
HelloWorld.createLogs();
}
}- The next step is to configure Log4j with the log4j-kafka-appender.
Appendersare in charge of delivering LogEvents to their destinations. TheKafkatopic is the destination in this situation. The logging level is set to warn in order to avoid repetitive logging. Because the appender is synchronous by default and will block until acknowledged by the Kafka server, it is wrapped with anAsyncappender to enable asynchronous logging. TheAsyncAppenderaccepts pointers to other appenders and writes LogEvents to them on a different thread.
XML
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="info" name="spring-boot-kafka-log" packages="com.reloadly">
<Appenders>
<Kafka name="Kafka" topic="my-topic-1">
<PatternLayout pattern="%date %message"/>
<Property name="bootstrap.servers">localhost:9092</Property>
</Kafka>
<Async name="Async">
<AppenderRef ref="Kafka"/>
</Async>
<Console name="stdout" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} %-5p [%-7t] %F:%L - %m%n"/>
</Console>
</Appenders>
<Loggers>
<Root level="INFO">
<AppenderRef ref="Kafka"/>
<AppenderRef ref="stdout"/>
</Root>
<Logger name="org.apache.kafka" level="WARN" /><!-- avoid recursive logging -->
</Loggers>
</Configuration>- Next, we'll process the logs that our simple Java application sends to the Kafka cluster. To do so, we'll construct a simple consumer class whose sole purpose is to listen to the
Kafkatopic where the logs are being streamed and process them. Simply print out the log messages on the console in this case.
Java
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class Consumer {
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("group.id", "test-group");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<> (properties);
List<String> topics = new ArrayList<>();
topics.add("my-topic-1");
kafkaConsumer.subscribe(topics);
try{
while (true){
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMinutes(20));
for (ConsumerRecord<String, String> message: records){
System.out.println(String.format("Topic - %s, Partition - %d, Value: %s", message.topic(), message.partition(), message.value()));
}
}
}catch (Exception e){
System.out.println(e.getMessage());
}finally {
kafkaConsumer.close();
}
}
}- Ensure that the Kafka container is up. Then, run Log4jAppenderApplication.Java and Consumer. Java.
You can find the project on GitHub.
kafka
Opinions expressed by DZone contributors are their own.