Enterprise Java

Apache Kafka Messaging System and Workflow

1. Introduction

This is an in-depth article related to the Apache Kafka messaging system and workflow. Apache Kafka was created in Linkedin. It is an open-source project from Apache from 2011. Kafka framework was developed in java and scala. It supports publish-subscribe messaging and is fault-tolerant. It can scale and is performant.

2. Apache Kafka

Apache Kafka is based on publish-subscribe messaging. It can process the high volume of messages and send them from source to destination. It is recommended for online and offline message consumption. The messages can be persisted on the disk to avoid data loss. ZooKeeper sync service is the foundation layer of Kafka. It can be integrated with Apache Storm and Spark for big data analysis. Kafka has many advantages like reliability, scalability, durability, and performance. Kafka is fault-tolerant and can be partitioned. It can be distributed and replicated. Scaling can be achieved without downtime. It has distributed logging and can handle a high volume of messages. Terabytes of messages can be processed and stored with good performance. It can be used with minimal (almost zero) downtime and data loss.

Kafka can be used for gathering operational monitoring metrics. You can use Kafka for the creation of centralized feeds related to aggregated statistics. It can also be used for the collection of logs from distributed services. The logs can be provided in a single format which can be standardized. Big data frameworks like Storm and Spark use Kafka to read from the topic for processing. The data processed can be written and published to another topic. You can use Kafka for streaming because it is durable. You can use Kafka for real-time data processing and streaming. It has low latency during the delivery of messages. It can guarantee message delivery as it is fault tolerant. You can process a huge number of messages to different consumers. It can handle 2 Million writes per second. You can store the message data to the disk in Kafka. The data is stored in the RAM (page cache) of the Operating system. Hence Kafka is efficient and can transfer message information from RAM to a network socket.

2.1 Prerequisites

Java 7 or 8 is required on the Linux, Windows, or Mac operating system.

2.2 Download

You can download Java 8 can be downloaded from the Oracle web site .  Kafka framework’s latest releases are available from this website.

2.3 Setup

You can set the environment variables for JAVA_HOME and PATH. They can be set as shown below:

Setup

JAVA_HOME="/desktop/jdk1.8.0_73"
export JAVA_HOME
PATH=$JAVA_HOME/bin:$PATH
export PATH

The environment variables for maven are set as below:

Maven Environment

JAVA_HOME=”/jboss/jdk1.8.0_73″
export M2_HOME=/users/bhagvan.kommadi/Desktop/apache-maven-3.6.1
export M2=$M2_HOME/bin
export PATH=$M2:$PATH

2.4 How to download and install Apache Kafka

Apache Kafka’s latest releases are available from the apache Kafka website. After downloading the zip file can be extracted to a folder.

To start the zookeeper, you can use the command below:

Zoo keeper

bin/zookeeper-server-start.sh config/zookeeper.properties

The output of the above command is shown as below:

Zoo keeper

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$ bin/zookeeper-server-start.sh config/zookeeper.properties
[2021-04-28 22:55:35,003] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,012] WARN config/zookeeper.properties is relative. Prepend ./ to indicate that you're sure! (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,034] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,034] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,040] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-04-28 22:55:35,041] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-04-28 22:55:35,041] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-04-28 22:55:35,041] WARN Either no config or no quorum defined in config, running  in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2021-04-28 22:55:35,050] INFO Log4j 1.2 jmx support found and enabled. (org.apache.zookeeper.jmx.ManagedUtil)
[2021-04-28 22:55:35,072] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,072] WARN config/zookeeper.properties is relative. Prepend ./ to indicate that you're sure! (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,072] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,072] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-04-28 22:55:35,073] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2021-04-28 22:55:35,081] INFO zookeeper.snapshot.trust.empty : false (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2021-04-28 22:55:35,108] INFO Server environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,108] INFO Server environment:host.name=localhost (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,108] INFO Server environment:java.version=1.8.0_101 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,108] INFO Server environment:java.vendor=Oracle Corporation (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,108] INFO Server environment:java.home=/Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/jre (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,109] INFO Server environment:java.class.path=/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/activation-1.1.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/aopalliance-repackaged-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/argparse4j-0.7.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/audience-annotations-0.5.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/commons-cli-1.4.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/commons-lang3-3.8.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-api-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-basic-auth-extension-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-file-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-json-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-mirror-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-mirror-client-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-runtime-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-transforms-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/hk2-api-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/hk2-locator-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/hk2-utils-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-annotations-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-core-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-databind-2.10.5.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-dataformat-csv-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-datatype-jdk8-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-jaxrs-base-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-jaxrs-json-provider-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-module-jaxb-annotations-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-module-paranamer-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-module-scala_2.12-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.activation-api-1.2.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.annotation-api-1.3.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.inject-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.validation-api-2.0.2.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.xml.bind-api-2.3.2.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/javassist-3.27.0-GA.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/javax.servlet-api-3.1.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/javax.ws.rs-api-2.1.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jaxb-api-2.3.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-client-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-common-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-container-servlet-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-container-servlet-core-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-hk2-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-media-jaxb-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-server-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-client-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-continuation-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-http-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-io-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-security-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-server-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-servlet-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-servlets-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-util-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-util-ajax-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jline-3.12.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jopt-simple-5.0.4.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-clients-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-log4j-appender-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-metadata-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-raft-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-shell-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-examples-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-scala_2.12-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-test-utils-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-tools-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka_2.12-2.8.0-sources.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka_2.12-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/log4j-1.2.17.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/lz4-java-1.7.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/maven-artifact-3.6.3.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/metrics-core-2.2.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-buffer-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-codec-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-common-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-handler-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-resolver-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-transport-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-transport-native-epoll-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-transport-native-unix-common-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/osgi-resource-locator-1.0.3.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/paranamer-2.8.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/plexus-utils-3.2.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/reflections-0.9.12.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/rocksdbjni-5.18.4.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-collection-compat_2.12-2.3.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-java8-compat_2.12-0.9.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-library-2.12.13.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-logging_2.12-3.9.2.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-reflect-2.12.13.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/slf4j-api-1.7.30.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/slf4j-log4j12-1.7.30.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/snappy-java-1.1.8.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/zookeeper-3.5.9.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/zookeeper-jute-3.5.9.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/zstd-jni-1.4.9-1.jar (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,110] INFO Server environment:java.library.path=/Users/bhagvan.kommadi/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:. (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:java.io.tmpdir=/var/folders/cr/0y892lq14qv7r24yl0gh0_dm0000gp/T/ (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:java.compiler= (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:os.name=Mac OS X (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:os.arch=x86_64 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:os.version=10.16 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:user.name=bhagvan.kommadi (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,111] INFO Server environment:user.home=/Users/bhagvan.kommadi (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,112] INFO Server environment:user.dir=/Users/bhagvan.kommadi/Desktop/kafka_2.12-2.8.0 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,112] INFO Server environment:os.memory.free=499MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,112] INFO Server environment:os.memory.max=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,112] INFO Server environment:os.memory.total=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,116] INFO minSessionTimeout set to 6000 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,116] INFO maxSessionTimeout set to 60000 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,118] INFO Created server with tickTime 3000 minSessionTimeout 6000 maxSessionTimeout 60000 datadir /tmp/zookeeper/version-2 snapdir /tmp/zookeeper/version-2 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-04-28 22:55:35,152] INFO Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory (org.apache.zookeeper.server.ServerCnxnFactory)
[2021-04-28 22:55:35,171] INFO Configuring NIO connection handler with 10s sessionless connection timeout, 1 selector thread(s), 8 worker threads, and 64 kB direct buffers. (org.apache.zookeeper.server.NIOServerCnxnFactory)
[2021-04-28 22:55:35,196] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
[2021-04-28 22:55:35,234] INFO zookeeper.snapshotSizeFactor = 0.33 (org.apache.zookeeper.server.ZKDatabase)
[2021-04-28 22:55:35,241] INFO Reading snapshot /tmp/zookeeper/version-2/snapshot.0 (org.apache.zookeeper.server.persistence.FileSnap)
[2021-04-28 22:55:35,412] INFO Snapshotting: 0x24 to /tmp/zookeeper/version-2/snapshot.24 (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2021-04-28 22:55:35,452] INFO PrepRequestProcessor (sid:0) started, reconfigEnabled=false (org.apache.zookeeper.server.PrepRequestProcessor)
[2021-04-28 22:55:35,464] INFO Using checkIntervalMs=60000 maxPerMinute=10000 (org.apache.zookeeper.server.ContainerManager)
[2021-04-28 22:55:37,412] INFO Creating new log file: log.25 (org.apache.zookeeper.server.persistence.FileTxnLog)

You can now start the apache kafka server using the command below:

Apache Kafka Server Run Command

bin/kafka-server-start.sh config/server.properties

The output of the above command is as follows:

Apache Kafka Server Output

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$ bin/kafka-server-start.sh config/server.properties
[2021-04-28 22:59:17,629] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2021-04-28 22:59:18,906] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2021-04-28 22:59:19,071] INFO Registered signal handlers for TERM, INT, HUP (org.apache.kafka.common.utils.LoggingSignalHandler)
[2021-04-28 22:59:19,081] INFO starting (kafka.server.KafkaServer)
[2021-04-28 22:59:19,082] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
[2021-04-28 22:59:19,124] INFO [ZooKeeperClient Kafka server] Initializing a new session to localhost:2181. (kafka.zookeeper.ZooKeeperClient)
[2021-04-28 22:59:19,136] INFO Client environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,137] INFO Client environment:host.name=localhost (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,137] INFO Client environment:java.version=1.8.0_101 (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,137] INFO Client environment:java.vendor=Oracle Corporation (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,137] INFO Client environment:java.home=/Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/jre (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,138] INFO Client environment:java.class.path=/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/activation-1.1.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/aopalliance-repackaged-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/argparse4j-0.7.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/audience-annotations-0.5.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/commons-cli-1.4.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/commons-lang3-3.8.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-api-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-basic-auth-extension-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-file-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-json-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-mirror-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-mirror-client-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-runtime-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/connect-transforms-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/hk2-api-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/hk2-locator-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/hk2-utils-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-annotations-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-core-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-databind-2.10.5.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-dataformat-csv-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-datatype-jdk8-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-jaxrs-base-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-jaxrs-json-provider-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-module-jaxb-annotations-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-module-paranamer-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jackson-module-scala_2.12-2.10.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.activation-api-1.2.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.annotation-api-1.3.5.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.inject-2.6.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.validation-api-2.0.2.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jakarta.xml.bind-api-2.3.2.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/javassist-3.27.0-GA.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/javax.servlet-api-3.1.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/javax.ws.rs-api-2.1.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jaxb-api-2.3.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-client-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-common-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-container-servlet-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-container-servlet-core-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-hk2-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-media-jaxb-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jersey-server-2.31.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-client-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-continuation-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-http-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-io-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-security-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-server-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-servlet-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-servlets-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-util-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jetty-util-ajax-9.4.39.v20210325.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jline-3.12.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/jopt-simple-5.0.4.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-clients-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-log4j-appender-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-metadata-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-raft-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-shell-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-examples-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-scala_2.12-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-streams-test-utils-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka-tools-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka_2.12-2.8.0-sources.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/kafka_2.12-2.8.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/log4j-1.2.17.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/lz4-java-1.7.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/maven-artifact-3.6.3.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/metrics-core-2.2.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-buffer-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-codec-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-common-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-handler-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-resolver-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-transport-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-transport-native-epoll-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/netty-transport-native-unix-common-4.1.62.Final.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/osgi-resource-locator-1.0.3.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/paranamer-2.8.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/plexus-utils-3.2.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/reflections-0.9.12.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/rocksdbjni-5.18.4.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-collection-compat_2.12-2.3.0.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-java8-compat_2.12-0.9.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-library-2.12.13.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-logging_2.12-3.9.2.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/scala-reflect-2.12.13.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/slf4j-api-1.7.30.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/slf4j-log4j12-1.7.30.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/snappy-java-1.1.8.1.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/zookeeper-3.5.9.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/zookeeper-jute-3.5.9.jar:/Users/bhagvan.kommadi/desktop/kafka_2.12-2.8.0/bin/../libs/zstd-jni-1.4.9-1.jar (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,142] INFO Client environment:java.library.path=/Users/bhagvan.kommadi/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:. (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,142] INFO Client environment:java.io.tmpdir=/var/folders/cr/0y892lq14qv7r24yl0gh0_dm0000gp/T/ (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,143] INFO Client environment:java.compiler= (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,143] INFO Client environment:os.name=Mac OS X (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,143] INFO Client environment:os.arch=x86_64 (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,143] INFO Client environment:os.version=10.16 (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,143] INFO Client environment:user.name=bhagvan.kommadi (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,143] INFO Client environment:user.home=/Users/bhagvan.kommadi (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,144] INFO Client environment:user.dir=/Users/bhagvan.kommadi/Desktop/kafka_2.12-2.8.0 (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,144] INFO Client environment:os.memory.free=973MB (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,144] INFO Client environment:os.memory.max=1024MB (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,144] INFO Client environment:os.memory.total=1024MB (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,182] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=18000 watcher=kafka.zookeeper.ZooKeeperClient$ZooKeeperClientWatcher$@799f10e1 (org.apache.zookeeper.ZooKeeper)
[2021-04-28 22:59:19,194] INFO jute.maxbuffer value is 4194304 Bytes (org.apache.zookeeper.ClientCnxnSocket)
[2021-04-28 22:59:19,205] INFO zookeeper.request.timeout value is 0. feature enabled= (org.apache.zookeeper.ClientCnxn)
[2021-04-28 22:59:19,209] INFO [ZooKeeperClient Kafka server] Waiting until connected. (kafka.zookeeper.ZooKeeperClient)
[2021-04-28 22:59:19,216] INFO Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2021-04-28 22:59:19,235] INFO Socket connection established, initiating session, client: /0:0:0:0:0:0:0:1:57462, server: localhost/0:0:0:0:0:0:0:1:2181 (org.apache.zookeeper.ClientCnxn)
[2021-04-28 22:59:19,262] INFO Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x100006c965b0000, negotiated timeout = 18000 (org.apache.zookeeper.ClientCnxn)
[2021-04-28 22:59:19,266] INFO [ZooKeeperClient Kafka server] Connected. (kafka.zookeeper.ZooKeeperClient)
[2021-04-28 22:59:19,476] INFO [feature-zk-node-event-process-thread]: Starting (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)
[2021-04-28 22:59:19,920] INFO Updated cache from existing  to latest FinalizedFeaturesAndEpoch(features=Features{}, epoch=0). (kafka.server.FinalizedFeatureCache)
[2021-04-28 22:59:19,927] INFO Cluster ID = kfLKbCV1RxKLc8esTApkrQ (kafka.server.KafkaServer)
[2021-04-28 22:59:20,083] INFO KafkaConfig values: 
	advertised.host.name = null
	advertised.listeners = null
	advertised.port = null
	alter.config.policy.class.name = null
	alter.log.dirs.replication.quota.window.num = 11
	alter.log.dirs.replication.quota.window.size.seconds = 1
	authorizer.class.name = 
	auto.create.topics.enable = true
	auto.leader.rebalance.enable = true
	background.threads = 10
	broker.heartbeat.interval.ms = 2000
	broker.id = 0
	broker.id.generation.enable = true
	broker.rack = null
	broker.session.timeout.ms = 9000
	client.quota.callback.class = null
	compression.type = producer
	connection.failed.authentication.delay.ms = 100
	connections.max.idle.ms = 600000
	connections.max.reauth.ms = 0
	control.plane.listener.name = null
	controlled.shutdown.enable = true
	controlled.shutdown.max.retries = 3
	controlled.shutdown.retry.backoff.ms = 5000
	controller.listener.names = null
	controller.quorum.append.linger.ms = 25
	controller.quorum.election.backoff.max.ms = 1000
	controller.quorum.election.timeout.ms = 1000
	controller.quorum.fetch.timeout.ms = 2000
	controller.quorum.request.timeout.ms = 2000
	controller.quorum.retry.backoff.ms = 20
	controller.quorum.voters = []
	controller.quota.window.num = 11
	controller.quota.window.size.seconds = 1
	controller.socket.timeout.ms = 30000
	create.topic.policy.class.name = null
	default.replication.factor = 1
	delegation.token.expiry.check.interval.ms = 3600000
	delegation.token.expiry.time.ms = 86400000
	delegation.token.master.key = null
	delegation.token.max.lifetime.ms = 604800000
	delegation.token.secret.key = null
	delete.records.purgatory.purge.interval.requests = 1
	delete.topic.enable = true
	fetch.max.bytes = 57671680
	fetch.purgatory.purge.interval.requests = 1000
	group.initial.rebalance.delay.ms = 0
	group.max.session.timeout.ms = 1800000
	group.max.size = 2147483647
	group.min.session.timeout.ms = 6000
	host.name = 
	initial.broker.registration.timeout.ms = 60000
	inter.broker.listener.name = null
	inter.broker.protocol.version = 2.8-IV1
	kafka.metrics.polling.interval.secs = 10
	kafka.metrics.reporters = []
	leader.imbalance.check.interval.seconds = 300
	leader.imbalance.per.broker.percentage = 10
	listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
	listeners = null
	log.cleaner.backoff.ms = 15000
	log.cleaner.dedupe.buffer.size = 134217728
	log.cleaner.delete.retention.ms = 86400000
	log.cleaner.enable = true
	log.cleaner.io.buffer.load.factor = 0.9
	log.cleaner.io.buffer.size = 524288
	log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308
	log.cleaner.max.compaction.lag.ms = 9223372036854775807
	log.cleaner.min.cleanable.ratio = 0.5
	log.cleaner.min.compaction.lag.ms = 0
	log.cleaner.threads = 1
	log.cleanup.policy = [delete]
	log.dir = /tmp/kafka-logs
	log.dirs = /tmp/kafka-logs
	log.flush.interval.messages = 9223372036854775807
	log.flush.interval.ms = null
	log.flush.offset.checkpoint.interval.ms = 60000
	log.flush.scheduler.interval.ms = 9223372036854775807
	log.flush.start.offset.checkpoint.interval.ms = 60000
	log.index.interval.bytes = 4096
	log.index.size.max.bytes = 10485760
	log.message.downconversion.enable = true
	log.message.format.version = 2.8-IV1
	log.message.timestamp.difference.max.ms = 9223372036854775807
	log.message.timestamp.type = CreateTime
	log.preallocate = false
	log.retention.bytes = -1
	log.retention.check.interval.ms = 300000
	log.retention.hours = 168
	log.retention.minutes = null
	log.retention.ms = null
	log.roll.hours = 168
	log.roll.jitter.hours = 0
	log.roll.jitter.ms = null
	log.roll.ms = null
	log.segment.bytes = 1073741824
	log.segment.delete.delay.ms = 60000
	max.connection.creation.rate = 2147483647
	max.connections = 2147483647
	max.connections.per.ip = 2147483647
	max.connections.per.ip.overrides = 
	max.incremental.fetch.session.cache.slots = 1000
	message.max.bytes = 1048588
	metadata.log.dir = null
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	min.insync.replicas = 1
	node.id = -1
	num.io.threads = 8
	num.network.threads = 3
	num.partitions = 1
	num.recovery.threads.per.data.dir = 1
	num.replica.alter.log.dirs.threads = null
	num.replica.fetchers = 1
	offset.metadata.max.bytes = 4096
	offsets.commit.required.acks = -1
	offsets.commit.timeout.ms = 5000
	offsets.load.buffer.size = 5242880
	offsets.retention.check.interval.ms = 600000
	offsets.retention.minutes = 10080
	offsets.topic.compression.codec = 0
	offsets.topic.num.partitions = 50
	offsets.topic.replication.factor = 1
	offsets.topic.segment.bytes = 104857600
	password.encoder.cipher.algorithm = AES/CBC/PKCS5Padding
	password.encoder.iterations = 4096
	password.encoder.key.length = 128
	password.encoder.keyfactory.algorithm = null
	password.encoder.old.secret = null
	password.encoder.secret = null
	port = 9092
	principal.builder.class = null
	process.roles = []
	producer.purgatory.purge.interval.requests = 1000
	queued.max.request.bytes = -1
	queued.max.requests = 500
	quota.consumer.default = 9223372036854775807
	quota.producer.default = 9223372036854775807
	quota.window.num = 11
	quota.window.size.seconds = 1
	replica.fetch.backoff.ms = 1000
	replica.fetch.max.bytes = 1048576
	replica.fetch.min.bytes = 1
	replica.fetch.response.max.bytes = 10485760
	replica.fetch.wait.max.ms = 500
	replica.high.watermark.checkpoint.interval.ms = 5000
	replica.lag.time.max.ms = 30000
	replica.selector.class = null
	replica.socket.receive.buffer.bytes = 65536
	replica.socket.timeout.ms = 30000
	replication.quota.window.num = 11
	replication.quota.window.size.seconds = 1
	request.timeout.ms = 30000
	reserved.broker.max.id = 1000
	sasl.client.callback.handler.class = null
	sasl.enabled.mechanisms = [GSSAPI]
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.principal.to.local.rules = [DEFAULT]
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism.controller.protocol = GSSAPI
	sasl.mechanism.inter.broker.protocol = GSSAPI
	sasl.server.callback.handler.class = null
	security.inter.broker.protocol = PLAINTEXT
	security.providers = null
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	socket.receive.buffer.bytes = 102400
	socket.request.max.bytes = 104857600
	socket.send.buffer.bytes = 102400
	ssl.cipher.suites = []
	ssl.client.auth = none
	ssl.enabled.protocols = [TLSv1.2]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.principal.mapping.rules = DEFAULT
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.abort.timed.out.transaction.cleanup.interval.ms = 10000
	transaction.max.timeout.ms = 900000
	transaction.remove.expired.transaction.cleanup.interval.ms = 3600000
	transaction.state.log.load.buffer.size = 5242880
	transaction.state.log.min.isr = 1
	transaction.state.log.num.partitions = 50
	transaction.state.log.replication.factor = 1
	transaction.state.log.segment.bytes = 104857600
	transactional.id.expiration.ms = 604800000
	unclean.leader.election.enable = false
	zookeeper.clientCnxnSocket = null
	zookeeper.connect = localhost:2181
	zookeeper.connection.timeout.ms = 18000
	zookeeper.max.in.flight.requests = 10
	zookeeper.session.timeout.ms = 18000
	zookeeper.set.acl = false
	zookeeper.ssl.cipher.suites = null
	zookeeper.ssl.client.enable = false
	zookeeper.ssl.crl.enable = false
	zookeeper.ssl.enabled.protocols = null
	zookeeper.ssl.endpoint.identification.algorithm = HTTPS
	zookeeper.ssl.keystore.location = null
	zookeeper.ssl.keystore.password = null
	zookeeper.ssl.keystore.type = null
	zookeeper.ssl.ocsp.enable = false
	zookeeper.ssl.protocol = TLSv1.2
	zookeeper.ssl.truststore.location = null
	zookeeper.ssl.truststore.password = null
	zookeeper.ssl.truststore.type = null
	zookeeper.sync.time.ms = 2000
 (kafka.server.KafkaConfig)
[2021-04-28 22:59:20,097] INFO KafkaConfig values: 
	advertised.host.name = null
	advertised.listeners = null
	advertised.port = null
	alter.config.policy.class.name = null
	alter.log.dirs.replication.quota.window.num = 11
	alter.log.dirs.replication.quota.window.size.seconds = 1
	authorizer.class.name = 
	auto.create.topics.enable = true
	auto.leader.rebalance.enable = true
	background.threads = 10
	broker.heartbeat.interval.ms = 2000
	broker.id = 0
	broker.id.generation.enable = true
	broker.rack = null
	broker.session.timeout.ms = 9000
	client.quota.callback.class = null
	compression.type = producer
	connection.failed.authentication.delay.ms = 100
	connections.max.idle.ms = 600000
	connections.max.reauth.ms = 0
	control.plane.listener.name = null
	controlled.shutdown.enable = true
	controlled.shutdown.max.retries = 3
	controlled.shutdown.retry.backoff.ms = 5000
	controller.listener.names = null
	controller.quorum.append.linger.ms = 25
	controller.quorum.election.backoff.max.ms = 1000
	controller.quorum.election.timeout.ms = 1000
	controller.quorum.fetch.timeout.ms = 2000
	controller.quorum.request.timeout.ms = 2000
	controller.quorum.retry.backoff.ms = 20
	controller.quorum.voters = []
	controller.quota.window.num = 11
	controller.quota.window.size.seconds = 1
	controller.socket.timeout.ms = 30000
	create.topic.policy.class.name = null
	default.replication.factor = 1
	delegation.token.expiry.check.interval.ms = 3600000
	delegation.token.expiry.time.ms = 86400000
	delegation.token.master.key = null
	delegation.token.max.lifetime.ms = 604800000
	delegation.token.secret.key = null
	delete.records.purgatory.purge.interval.requests = 1
	delete.topic.enable = true
	fetch.max.bytes = 57671680
	fetch.purgatory.purge.interval.requests = 1000
	group.initial.rebalance.delay.ms = 0
	group.max.session.timeout.ms = 1800000
	group.max.size = 2147483647
	group.min.session.timeout.ms = 6000
	host.name = 
	initial.broker.registration.timeout.ms = 60000
	inter.broker.listener.name = null
	inter.broker.protocol.version = 2.8-IV1
	kafka.metrics.polling.interval.secs = 10
	kafka.metrics.reporters = []
	leader.imbalance.check.interval.seconds = 300
	leader.imbalance.per.broker.percentage = 10
	listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
	listeners = null
	log.cleaner.backoff.ms = 15000
	log.cleaner.dedupe.buffer.size = 134217728
	log.cleaner.delete.retention.ms = 86400000
	log.cleaner.enable = true
	log.cleaner.io.buffer.load.factor = 0.9
	log.cleaner.io.buffer.size = 524288
	log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308
	log.cleaner.max.compaction.lag.ms = 9223372036854775807
	log.cleaner.min.cleanable.ratio = 0.5
	log.cleaner.min.compaction.lag.ms = 0
	log.cleaner.threads = 1
	log.cleanup.policy = [delete]
	log.dir = /tmp/kafka-logs
	log.dirs = /tmp/kafka-logs
	log.flush.interval.messages = 9223372036854775807
	log.flush.interval.ms = null
	log.flush.offset.checkpoint.interval.ms = 60000
	log.flush.scheduler.interval.ms = 9223372036854775807
	log.flush.start.offset.checkpoint.interval.ms = 60000
	log.index.interval.bytes = 4096
	log.index.size.max.bytes = 10485760
	log.message.downconversion.enable = true
	log.message.format.version = 2.8-IV1
	log.message.timestamp.difference.max.ms = 9223372036854775807
	log.message.timestamp.type = CreateTime
	log.preallocate = false
	log.retention.bytes = -1
	log.retention.check.interval.ms = 300000
	log.retention.hours = 168
	log.retention.minutes = null
	log.retention.ms = null
	log.roll.hours = 168
	log.roll.jitter.hours = 0
	log.roll.jitter.ms = null
	log.roll.ms = null
	log.segment.bytes = 1073741824
	log.segment.delete.delay.ms = 60000
	max.connection.creation.rate = 2147483647
	max.connections = 2147483647
	max.connections.per.ip = 2147483647
	max.connections.per.ip.overrides = 
	max.incremental.fetch.session.cache.slots = 1000
	message.max.bytes = 1048588
	metadata.log.dir = null
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	min.insync.replicas = 1
	node.id = -1
	num.io.threads = 8
	num.network.threads = 3
	num.partitions = 1
	num.recovery.threads.per.data.dir = 1
	num.replica.alter.log.dirs.threads = null
	num.replica.fetchers = 1
	offset.metadata.max.bytes = 4096
	offsets.commit.required.acks = -1
	offsets.commit.timeout.ms = 5000
	offsets.load.buffer.size = 5242880
	offsets.retention.check.interval.ms = 600000
	offsets.retention.minutes = 10080
	offsets.topic.compression.codec = 0
	offsets.topic.num.partitions = 50
	offsets.topic.replication.factor = 1
	offsets.topic.segment.bytes = 104857600
	password.encoder.cipher.algorithm = AES/CBC/PKCS5Padding
	password.encoder.iterations = 4096
	password.encoder.key.length = 128
	password.encoder.keyfactory.algorithm = null
	password.encoder.old.secret = null
	password.encoder.secret = null
	port = 9092
	principal.builder.class = null
	process.roles = []
	producer.purgatory.purge.interval.requests = 1000
	queued.max.request.bytes = -1
	queued.max.requests = 500
	quota.consumer.default = 9223372036854775807
	quota.producer.default = 9223372036854775807
	quota.window.num = 11
	quota.window.size.seconds = 1
	replica.fetch.backoff.ms = 1000
	replica.fetch.max.bytes = 1048576
	replica.fetch.min.bytes = 1
	replica.fetch.response.max.bytes = 10485760
	replica.fetch.wait.max.ms = 500
	replica.high.watermark.checkpoint.interval.ms = 5000
	replica.lag.time.max.ms = 30000
	replica.selector.class = null
	replica.socket.receive.buffer.bytes = 65536
	replica.socket.timeout.ms = 30000
	replication.quota.window.num = 11
	replication.quota.window.size.seconds = 1
	request.timeout.ms = 30000
	reserved.broker.max.id = 1000
	sasl.client.callback.handler.class = null
	sasl.enabled.mechanisms = [GSSAPI]
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.principal.to.local.rules = [DEFAULT]
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism.controller.protocol = GSSAPI
	sasl.mechanism.inter.broker.protocol = GSSAPI
	sasl.server.callback.handler.class = null
	security.inter.broker.protocol = PLAINTEXT
	security.providers = null
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	socket.receive.buffer.bytes = 102400
	socket.request.max.bytes = 104857600
	socket.send.buffer.bytes = 102400
	ssl.cipher.suites = []
	ssl.client.auth = none
	ssl.enabled.protocols = [TLSv1.2]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.principal.mapping.rules = DEFAULT
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.abort.timed.out.transaction.cleanup.interval.ms = 10000
	transaction.max.timeout.ms = 900000
	transaction.remove.expired.transaction.cleanup.interval.ms = 3600000
	transaction.state.log.load.buffer.size = 5242880
	transaction.state.log.min.isr = 1
	transaction.state.log.num.partitions = 50
	transaction.state.log.replication.factor = 1
	transaction.state.log.segment.bytes = 104857600
	transactional.id.expiration.ms = 604800000
	unclean.leader.election.enable = false
	zookeeper.clientCnxnSocket = null
	zookeeper.connect = localhost:2181
	zookeeper.connection.timeout.ms = 18000
	zookeeper.max.in.flight.requests = 10
	zookeeper.session.timeout.ms = 18000
	zookeeper.set.acl = false
	zookeeper.ssl.cipher.suites = null
	zookeeper.ssl.client.enable = false
	zookeeper.ssl.crl.enable = false
	zookeeper.ssl.enabled.protocols = null
	zookeeper.ssl.endpoint.identification.algorithm = HTTPS
	zookeeper.ssl.keystore.location = null
	zookeeper.ssl.keystore.password = null
	zookeeper.ssl.keystore.type = null
	zookeeper.ssl.ocsp.enable = false
	zookeeper.ssl.protocol = TLSv1.2
	zookeeper.ssl.truststore.location = null
	zookeeper.ssl.truststore.password = null
	zookeeper.ssl.truststore.type = null
	zookeeper.sync.time.ms = 2000
 (kafka.server.KafkaConfig)
[2021-04-28 22:59:20,179] INFO [ThrottledChannelReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-04-28 22:59:20,180] INFO [ThrottledChannelReaper-Produce]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-04-28 22:59:20,181] INFO [ThrottledChannelReaper-Request]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-04-28 22:59:20,185] INFO [ThrottledChannelReaper-ControllerMutation]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-04-28 22:59:20,268] INFO Loading logs from log dirs ArrayBuffer(/tmp/kafka-logs) (kafka.log.LogManager)
[2021-04-28 22:59:20,273] INFO Skipping recovery for all logs in /tmp/kafka-logs since clean shutdown file was found (kafka.log.LogManager)
[2021-04-28 22:59:20,460] INFO [Log partition=quickstart-events-0, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-04-28 22:59:20,479] INFO Completed load of Log(dir=/tmp/kafka-logs/quickstart-events-0, topic=quickstart-events, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 182ms (1/1 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-04-28 22:59:20,482] INFO Loaded 1 logs in 213ms. (kafka.log.LogManager)
[2021-04-28 22:59:20,484] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
[2021-04-28 22:59:20,485] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
[2021-04-28 22:59:21,742] INFO Updated connection-accept-rate max connection creation rate to 2147483647 (kafka.network.ConnectionQuotas)
[2021-04-28 22:59:21,751] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.Acceptor)
[2021-04-28 22:59:21,829] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Created data-plane acceptor and processors for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-04-28 22:59:21,924] INFO [broker-0-to-controller-send-thread]: Starting (kafka.server.BrokerToControllerRequestThread)
[2021-04-28 22:59:21,966] INFO [ExpirationReaper-0-Produce]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:21,972] INFO [ExpirationReaper-0-Fetch]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:21,972] INFO [ExpirationReaper-0-DeleteRecords]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:21,975] INFO [ExpirationReaper-0-ElectLeader]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:22,028] INFO [LogDirFailureHandler]: Starting (kafka.server.ReplicaManager$LogDirFailureHandler)
[2021-04-28 22:59:22,123] INFO Creating /brokers/ids/0 (is it secure? false) (kafka.zk.KafkaZkClient)
[2021-04-28 22:59:22,159] INFO Stat of the created znode at /brokers/ids/0 is: 53,53,1619630962148,1619630962148,1,0,0,72058060416942080,202,0,53
 (kafka.zk.KafkaZkClient)
[2021-04-28 22:59:22,161] INFO Registered broker 0 at path /brokers/ids/0 with addresses: PLAINTEXT://localhost:9092, czxid (broker epoch): 53 (kafka.zk.KafkaZkClient)
[2021-04-28 22:59:22,380] INFO [ExpirationReaper-0-topic]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:22,395] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:22,402] INFO [ExpirationReaper-0-Rebalance]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:22,462] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.group.GroupCoordinator)
[2021-04-28 22:59:22,471] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.group.GroupCoordinator)
[2021-04-28 22:59:22,533] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:1000,blockEndProducerId:1999) by writing to Zk with path version 2 (kafka.coordinator.transaction.ProducerIdManager)
[2021-04-28 22:59:22,534] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2021-04-28 22:59:22,543] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2021-04-28 22:59:22,543] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2021-04-28 22:59:22,612] INFO [ExpirationReaper-0-AlterAcls]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-04-28 22:59:22,687] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2021-04-28 22:59:22,737] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Starting socket server acceptors and processors (kafka.network.SocketServer)
[2021-04-28 22:59:22,753] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-04-28 22:59:22,759] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started socket server acceptors and processors (kafka.network.SocketServer)
[2021-04-28 22:59:22,785] INFO Kafka version: 2.8.0 (org.apache.kafka.common.utils.AppInfoParser)
[2021-04-28 22:59:22,786] INFO Kafka commitId: ebb1d6e21cc92130 (org.apache.kafka.common.utils.AppInfoParser)
[2021-04-28 22:59:22,786] INFO Kafka startTimeMs: 1619630962759 (org.apache.kafka.common.utils.AppInfoParser)
[2021-04-28 22:59:22,790] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
[2021-04-28 22:59:22,873] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions Set(quickstart-events-0) (kafka.server.ReplicaFetcherManager)
[2021-04-28 22:59:22,898] INFO [broker-0-to-controller-send-thread]: Recorded new controller, from now on will use broker localhost:9092 (id: 0 rack: null) (kafka.server.BrokerToControllerRequestThread)
[2021-04-28 22:59:22,903] INFO [Partition quickstart-events-0 broker=0] Log loaded for partition quickstart-events-0 with initial high watermark 0 (kafka.cluster.Partition)

2.5 What is messaging system ?

Kafka has a distributed messaging system. A messaging system can move messages or data from source to destination. The movement of messages is independent of the messages and data. The messaging and the delivery is based on reliable message queues in the messaging system. You can send asynchronously between source and destination using queues. Messaging systems support point to point and publish-subscribe messaging patterns.

2.6 Kafka Workflow

Apache Kafka workflow helps in managing topics, partitions, producers, and consumers. You can have a set of topics that can be partitioned. A partition is a set of messages which are order and are identified by the offset. In a Kafka cluster, the union of partitions of data is disjoint. Kafka workflow supports both publish-subscribe and queuing message patterns.

2.6.1 Point to Point Messaging workflow

In Kafka, You can send messages using a queue and store them. Consumers can consume the message data from the queue. In a message queue, only one consumer takes or consumes a specific message. After the consumer takes the message, that specific message will not be there in the queue. There are many examples for point-to-point messaging like order processing, P2P messaging, and others. A topic can be used for a set of consumers who belong to a group.

You can create a message queue and send events to the message queue as shown below in the code snippet:

Apache Kafka Server Message Queue

BlockingQueue blockingQueue = new  LinkedBlockingQueue(1000);  
        Client client = Messagingclient(blockingQueue);  
        client.connect(); 
        KafkaProducer producer=createKafkaMessageProducer();  
public KafkaProducer createKafkaMessageProducer(){  
    
        String bootstrapServers="127.0.0.1:9092";  
        Properties properties= new Properties();  
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,    bootstrapServers);  
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());  
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());  
  
        KafkaProducer producer = new KafkaProducer(properties);  
        return producer;  
  
    }  

2.6.2 Publish-Subscribe Messaging workflow

Kafka is based on publish and subscribe messaging patterns. In this pattern, you can persist the messages in the topic. Multiple consumers can consume from the topic the same message. Message producers are referred to as publishers and consumers as subscribers. Kafka workflow can be used to configure the sending of messages at regular intervals by producers. Messages are stored by the Kafka broker. Messages can be shared in the topic-specific partitions. Messages can be stored sequentially in different partitions by their order. The consumer receives the message based on the topic subscription. After the message is consumed, the zookeeper saves the offset in the ensemble. You can configure the interval for consuming the messages on Kafka workflow. You can have acknowledgments sent to the broker from the consumer. Kafka broker updates the Zookeeper after getting the acknowledgment.

To create a topic on the apache kafka server, you can use the command below:

Apache Kafka Server Topic creation

bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

The output of the above command is as follows:

Apache Kafka Server Topic creation output

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events	TopicId: rBGQvg4kTcKK_Kf0IxP-lw	PartitionCount: 1	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: quickstart-events	Partition: 0	Leader: 0	Replicas: 0	Isr: 0

Now let us create events to be posted on the topic using the command below:

Apache Kafka Server Topic – Posting events

bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
>event 1
>event 2

To read the events on the topic, you can use the command below:

Apache Kafka Server Topic – Reading events

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$ bin/kafka-console-consumsh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
event 1
event 2

2.7 Cluster Architecture

Kafka messaging system supports clustering. The cluster can have more than one broker and supports load balancing. The brokers are stateless and the state is maintained by the Zookeeper. Terabytes of messages can be processed by the broker and multiple instances of broker can be managed by the ZooKeeper. Broker leader can be selected by the ZooKeeper. Zookeeper service helps in sending notifications about the broker to consumers and producers. Notification can be of presence or failure of the broker. You can push messages to brokers using producers. When a new broker is detected, producers send the messages to the broker. Broker is selected as a leader when it is responsible for the reads and writes happening in a partition. The follower is the broker which listens to the leader’s instructions.

A typical Kafka architecture consists of brokers, producers, consumers, and Zookeeper. Brokers are part of the cluster and load balancing is supported in Kafka. The state of the brokers is stored in Zookeeper. Zookeeper manages the brokers and notifies the producer and consumer regarding brokers.

3. Download the Source Code

Download
You can download the full source code of this example here: Apache Kafka Messaging System and Workflow

Bhagvan Kommadi

Bhagvan Kommadi is the Founder of Architect Corner & has around 20 years’ experience in the industry, ranging from large scale enterprise development to helping incubate software product start-ups. He has done Masters in Industrial Systems Engineering at Georgia Institute of Technology (1997) and Bachelors in Aerospace Engineering from Indian Institute of Technology, Madras (1993). He is member of IFX forum,Oracle JCP and participant in Java Community Process. He founded Quantica Computacao, the first quantum computing startup in India. Markets and Markets have positioned Quantica Computacao in ‘Emerging Companies’ section of Quantum Computing quadrants. Bhagvan has engineered and developed simulators and tools in the area of quantum technology using IBM Q, Microsoft Q# and Google QScript. He has reviewed the Manning book titled : "Machine Learning with TensorFlow”. He is also the author of Packt Publishing book - "Hands-On Data Structures and Algorithms with Go".He is member of IFX forum,Oracle JCP and participant in Java Community Process. He is member of the MIT Technology Review Global Panel.
Subscribe
Notify of
guest

This site uses Akismet to reduce spam. Learn how your comment data is processed.

0 Comments
Oldest
Newest Most Voted
Inline Feedbacks
View all comments
Back to top button