Enterprise Java

Apache Kafka Integration With Spark

1. Introduction

This is an in-depth article related to Apache Kafka and Spark Integration. Apache Kafka is an Apache open-source project. It was initially created on Linkedin. Kafka framework was created in java and scala. It supports publish-subscribe messaging and is fault-tolerant. It is scalable and performs for high-volume messaging. Zookeeper is the basic component that manages the Apache Kafka Server. Kafka has features related to reliability, scalability, performance, distributed logging, and durability. Spark is used for big data processing and executing machine learning algorithms.

2. Apache Kafka – Spark Integration

2.1 Prerequisites

Java 7 or 8 is required on the linux, windows or mac operating system. Maven 3.6.1 is required for building the kafka and spark application.

2.2 Download

You can download Java 8 can be downloaded from the Oracle web site . Apache Maven 3.6.1 can be downloaded from Apache 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 Output

[2021-06-04 01:19:44,763] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,767] WARN config/zookeeper.properties is relative. Prepend ./ to indicate that you're sure! (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,789] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,789] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,795] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-06-04 01:19:44,795] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-06-04 01:19:44,795] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-06-04 01:19:44,796] WARN Either no config or no quorum defined in config, running  in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2021-06-04 01:19:44,804] INFO Log4j 1.2 jmx support found and enabled. (org.apache.zookeeper.jmx.ManagedUtil)
[2021-06-04 01:19:44,824] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,824] WARN config/zookeeper.properties is relative. Prepend ./ to indicate that you're sure! (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,824] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,825] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-06-04 01:19:44,825] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2021-06-04 01:19:44,832] INFO zookeeper.snapshot.trust.empty : false (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2021-06-04 01:19:44,858] INFO Server environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,858] INFO Server environment:host.name=localhost (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,858] INFO Server environment:java.version=1.8.0_101 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,858] INFO Server environment:java.vendor=Oracle Corporation (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,858] INFO Server environment:java.home=/Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/jre (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,859] 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-06-04 01:19:44,859] 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-06-04 01:19:44,859] INFO Server environment:java.io.tmpdir=/var/folders/cr/0y892lq14qv7r24yl0gh0_dm0000gp/T/ (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:java.compiler= (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:os.name=Mac OS X (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:os.arch=x86_64 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:os.version=10.16 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:user.name=bhagvan.kommadi (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:user.home=/Users/bhagvan.kommadi (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:user.dir=/Users/bhagvan.kommadi/Desktop/kafka_2.12-2.8.0 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:os.memory.free=499MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:os.memory.max=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,860] INFO Server environment:os.memory.total=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,863] INFO minSessionTimeout set to 6000 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,863] INFO maxSessionTimeout set to 60000 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-06-04 01:19:44,864] 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-06-04 01:19:44,896] INFO Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory (org.apache.zookeeper.server.ServerCnxnFactory)
[2021-06-04 01:19:44,913] 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-06-04 01:19:44,938] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
[2021-06-04 01:19:44,968] INFO zookeeper.snapshotSizeFactor = 0.33 (org.apache.zookeeper.server.ZKDatabase)
[2021-06-04 01:19:44,985] INFO Reading snapshot /tmp/zookeeper/version-2/snapshot.c0 (org.apache.zookeeper.server.persistence.FileSnap)
[2021-06-04 01:19:45,144] INFO Snapshotting: 0x10a to /tmp/zookeeper/version-2/snapshot.10a (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2021-06-04 01:19:45,176] INFO PrepRequestProcessor (sid:0) started, reconfigEnabled=false (org.apache.zookeeper.server.PrepRequestProcessor)
[2021-06-04 01:19:45,183] INFO Using checkIntervalMs=60000 maxPerMinute=10000 (org.apache.zookeeper.server.ContainerManager)
[2021-06-04 01:19:57,850] INFO Creating new log file: log.10b (org.apache.zookeeper.server.persistence.FileTxnLog)
[2021-06-04 01:20:03,791] INFO Expiring session 0x10000d006010002, timeout of 18000ms exceeded (org.apache.zookeeper.server.ZooKeeperServer)

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 shown as below:

Apache Kafka Server Output

[2021-06-04 01:20:26,929] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2021-06-04 01:20:27,762] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2021-06-04 01:20:27,881] INFO Registered signal handlers for TERM, INT, HUP (org.apache.kafka.common.utils.LoggingSignalHandler)
[2021-06-04 01:20:27,887] INFO starting (kafka.server.KafkaServer)
[2021-06-04 01:20:27,888] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
[2021-06-04 01:20:27,918] INFO [ZooKeeperClient Kafka server] Initializing a new session to localhost:2181. (kafka.zookeeper.ZooKeeperClient)
[2021-06-04 01:20:27,929] INFO Client environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,929] INFO Client environment:host.name=localhost (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,929] INFO Client environment:java.version=1.8.0_101 (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,929] INFO Client environment:java.vendor=Oracle Corporation (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,929] INFO Client environment:java.home=/Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/jre (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,930] 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-06-04 01:20:27,930] 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-06-04 01:20:27,931] INFO Client environment:java.io.tmpdir=/var/folders/cr/0y892lq14qv7r24yl0gh0_dm0000gp/T/ (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:java.compiler= (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:os.name=Mac OS X (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:os.arch=x86_64 (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:os.version=10.16 (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:user.name=bhagvan.kommadi (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:user.home=/Users/bhagvan.kommadi (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:user.dir=/Users/bhagvan.kommadi/Desktop/kafka_2.12-2.8.0 (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,931] INFO Client environment:os.memory.free=973MB (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,932] INFO Client environment:os.memory.max=1024MB (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,932] INFO Client environment:os.memory.total=1024MB (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,952] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=18000 watcher=kafka.zookeeper.ZooKeeperClient$ZooKeeperClientWatcher$@799f10e1 (org.apache.zookeeper.ZooKeeper)
[2021-06-04 01:20:27,962] INFO jute.maxbuffer value is 4194304 Bytes (org.apache.zookeeper.ClientCnxnSocket)
[2021-06-04 01:20:27,969] INFO zookeeper.request.timeout value is 0. feature enabled= (org.apache.zookeeper.ClientCnxn)
[2021-06-04 01:20:27,972] INFO [ZooKeeperClient Kafka server] Waiting until connected. (kafka.zookeeper.ZooKeeperClient)
[2021-06-04 01:20:27,976] INFO Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2021-06-04 01:20:27,994] INFO Socket connection established, initiating session, client: /127.0.0.1:52537, server: localhost/127.0.0.1:2181 (org.apache.zookeeper.ClientCnxn)
[2021-06-04 01:20:28,002] INFO Session establishment complete on server localhost/127.0.0.1:2181, sessionid = 0x100018ed7060001, negotiated timeout = 18000 (org.apache.zookeeper.ClientCnxn)
[2021-06-04 01:20:28,006] INFO [ZooKeeperClient Kafka server] Connected. (kafka.zookeeper.ZooKeeperClient)
[2021-06-04 01:20:28,107] INFO [feature-zk-node-event-process-thread]: Starting (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)
[2021-06-04 01:20:28,336] INFO Updated cache from existing  to latest FinalizedFeaturesAndEpoch(features=Features{}, epoch=0). (kafka.server.FinalizedFeatureCache)
[2021-06-04 01:20:28,343] INFO Cluster ID = s6HNO0aXTMC5pyby0hV2kw (kafka.server.KafkaServer)
[2021-06-04 01:20:28,472] 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-06-04 01:20:28,483] 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-06-04 01:20:28,550] INFO [ThrottledChannelReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-06-04 01:20:28,550] INFO [ThrottledChannelReaper-Produce]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-06-04 01:20:28,552] INFO [ThrottledChannelReaper-Request]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-06-04 01:20:28,554] INFO [ThrottledChannelReaper-ControllerMutation]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-06-04 01:20:28,623] INFO Loading logs from log dirs ArrayBuffer(/tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:28,628] INFO Skipping recovery for all logs in /tmp/kafka-logs since clean shutdown file was found (kafka.log.LogManager)
[2021-06-04 01:20:28,805] INFO [Log partition=__consumer_offsets-9, dir=/tmp/kafka-logs] Loading producer state till offset 547 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:28,862] INFO [ProducerStateManager partition=__consumer_offsets-9] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/__consumer_offsets-9/00000000000000000547.snapshot,547)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:28,921] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-9, topic=__consumer_offsets, partition=9, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=547) with 1 segments in 263ms (1/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:28,928] INFO [Log partition=__consumer_offsets-0, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:28,935] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-0, topic=__consumer_offsets, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 13ms (2/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:28,941] INFO [Log partition=__consumer_offsets-7, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:28,944] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-7, topic=__consumer_offsets, partition=7, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 9ms (3/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:28,952] INFO [Log partition=__consumer_offsets-31, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:28,955] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-31, topic=__consumer_offsets, partition=31, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 10ms (4/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:28,962] INFO [Log partition=__consumer_offsets-36, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:28,966] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-36, topic=__consumer_offsets, partition=36, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 10ms (5/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:28,991] INFO [Log partition=__consumer_offsets-38, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:28,998] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-38, topic=__consumer_offsets, partition=38, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 30ms (6/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,005] INFO [Log partition=__consumer_offsets-6, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,016] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-6, topic=__consumer_offsets, partition=6, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 18ms (7/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,024] INFO [Log partition=__consumer_offsets-1, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,028] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-1, topic=__consumer_offsets, partition=1, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 11ms (8/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,036] INFO [Log partition=__consumer_offsets-8, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,039] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-8, topic=__consumer_offsets, partition=8, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 10ms (9/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,056] INFO [Log partition=__consumer_offsets-39, dir=/tmp/kafka-logs] Loading producer state till offset 28 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,056] INFO [ProducerStateManager partition=__consumer_offsets-39] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/__consumer_offsets-39/00000000000000000028.snapshot,28)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:29,061] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-39, topic=__consumer_offsets, partition=39, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=28) with 1 segments in 21ms (10/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,068] INFO [Log partition=__consumer_offsets-37, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,072] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-37, topic=__consumer_offsets, partition=37, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 11ms (11/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,078] INFO [Log partition=__consumer_offsets-30, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,080] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-30, topic=__consumer_offsets, partition=30, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 8ms (12/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,090] INFO [Log partition=messages-0, dir=/tmp/kafka-logs] Loading producer state till offset 168 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,090] INFO [ProducerStateManager partition=messages-0] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/messages-0/00000000000000000168.snapshot,168)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:29,096] INFO Completed load of Log(dir=/tmp/kafka-logs/messages-0, topic=messages, partition=0, highWatermark=146, lastStableOffset=146, logStartOffset=146, logEndOffset=168) with 1 segments in 16ms (13/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,102] INFO [Log partition=__consumer_offsets-12, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,103] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-12, topic=__consumer_offsets, partition=12, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (14/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,110] INFO [Log partition=__consumer_offsets-15, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,113] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-15, topic=__consumer_offsets, partition=15, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 9ms (15/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,120] INFO [Log partition=__consumer_offsets-23, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,123] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-23, topic=__consumer_offsets, partition=23, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 9ms (16/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,128] INFO [Log partition=__consumer_offsets-24, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,130] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-24, topic=__consumer_offsets, partition=24, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (17/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,134] INFO [Log partition=__consumer_offsets-48, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,136] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-48, topic=__consumer_offsets, partition=48, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (18/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,141] INFO [Log partition=__consumer_offsets-41, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,143] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-41, topic=__consumer_offsets, partition=41, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (19/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,148] INFO [Log partition=__consumer_offsets-46, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,150] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-46, topic=__consumer_offsets, partition=46, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 8ms (20/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,156] INFO [Log partition=__consumer_offsets-25, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,157] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-25, topic=__consumer_offsets, partition=25, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (21/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,162] INFO [Log partition=__consumer_offsets-22, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,164] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-22, topic=__consumer_offsets, partition=22, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (22/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,169] INFO [Log partition=__consumer_offsets-14, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,170] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-14, topic=__consumer_offsets, partition=14, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (23/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,175] INFO [Log partition=__consumer_offsets-13, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,178] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-13, topic=__consumer_offsets, partition=13, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (24/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,182] INFO [Log partition=__consumer_offsets-47, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,184] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-47, topic=__consumer_offsets, partition=47, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (25/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,189] INFO [Log partition=NewTopic-0, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,191] INFO Completed load of Log(dir=/tmp/kafka-logs/NewTopic-0, topic=NewTopic, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (26/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,198] INFO [Log partition=__consumer_offsets-40, dir=/tmp/kafka-logs] Loading producer state till offset 7 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,198] INFO [ProducerStateManager partition=__consumer_offsets-40] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/__consumer_offsets-40/00000000000000000007.snapshot,7)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:29,201] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-40, topic=__consumer_offsets, partition=40, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=7) with 1 segments in 10ms (27/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,205] INFO [Log partition=__consumer_offsets-49, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,208] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-49, topic=__consumer_offsets, partition=49, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (28/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,215] INFO [Log partition=new_topic-0, dir=/tmp/kafka-logs] Loading producer state till offset 4 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,215] INFO [ProducerStateManager partition=new_topic-0] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/new_topic-0/00000000000000000004.snapshot,4)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:29,217] INFO Completed load of Log(dir=/tmp/kafka-logs/new_topic-0, topic=new_topic, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=4) with 1 segments in 9ms (29/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,222] INFO [Log partition=__consumer_offsets-35, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,225] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-35, topic=__consumer_offsets, partition=35, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (30/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,230] INFO [Log partition=__consumer_offsets-32, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,232] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-32, topic=__consumer_offsets, partition=32, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (31/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,239] INFO [Log partition=twitter_topic-0, dir=/tmp/kafka-logs] Loading producer state till offset 8 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,239] INFO [ProducerStateManager partition=twitter_topic-0] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/twitter_topic-0/00000000000000000008.snapshot,8)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:29,241] INFO Completed load of Log(dir=/tmp/kafka-logs/twitter_topic-0, topic=twitter_topic, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=8) with 1 segments in 9ms (32/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,245] INFO [Log partition=__consumer_offsets-4, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,247] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-4, topic=__consumer_offsets, partition=4, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (33/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,252] INFO [Log partition=__consumer_offsets-3, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,254] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-3, topic=__consumer_offsets, partition=3, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 7ms (34/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,259] INFO [Log partition=__consumer_offsets-33, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,260] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-33, topic=__consumer_offsets, partition=33, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (35/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,265] INFO [Log partition=__consumer_offsets-34, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,266] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-34, topic=__consumer_offsets, partition=34, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (36/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,271] INFO [Log partition=__consumer_offsets-2, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,272] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-2, topic=__consumer_offsets, partition=2, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (37/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,276] INFO [Log partition=__consumer_offsets-5, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,277] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-5, topic=__consumer_offsets, partition=5, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (38/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,281] INFO [Log partition=__consumer_offsets-45, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,282] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-45, topic=__consumer_offsets, partition=45, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 4ms (39/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,287] INFO [Log partition=__consumer_offsets-42, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,288] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-42, topic=__consumer_offsets, partition=42, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (40/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,293] INFO [Log partition=__consumer_offsets-29, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,294] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-29, topic=__consumer_offsets, partition=29, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (41/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,298] INFO [Log partition=__consumer_offsets-16, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,299] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-16, topic=__consumer_offsets, partition=16, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 4ms (42/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,304] INFO [Log partition=__consumer_offsets-11, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,305] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-11, topic=__consumer_offsets, partition=11, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (43/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,311] INFO [Log partition=__consumer_offsets-18, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,312] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-18, topic=__consumer_offsets, partition=18, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (44/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,316] INFO [Log partition=__consumer_offsets-27, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,317] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-27, topic=__consumer_offsets, partition=27, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (45/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,321] INFO [Log partition=__consumer_offsets-20, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,322] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-20, topic=__consumer_offsets, partition=20, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (46/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,327] INFO [Log partition=__consumer_offsets-43, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,328] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-43, topic=__consumer_offsets, partition=43, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 6ms (47/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,332] INFO [Log partition=__consumer_offsets-44, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,333] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-44, topic=__consumer_offsets, partition=44, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (48/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,337] INFO [Log partition=__consumer_offsets-21, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,338] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-21, topic=__consumer_offsets, partition=21, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (49/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,342] INFO [Log partition=__consumer_offsets-19, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,343] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-19, topic=__consumer_offsets, partition=19, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (50/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,346] INFO [Log partition=__consumer_offsets-26, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,347] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-26, topic=__consumer_offsets, partition=26, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 3ms (51/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,355] INFO [Log partition=__consumer_offsets-10, dir=/tmp/kafka-logs] Loading producer state till offset 54 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,356] INFO [ProducerStateManager partition=__consumer_offsets-10] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/__consumer_offsets-10/00000000000000000054.snapshot,54)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:29,357] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-10, topic=__consumer_offsets, partition=10, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=54) with 1 segments in 9ms (52/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,362] INFO [Log partition=__consumer_offsets-28, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,363] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-28, topic=__consumer_offsets, partition=28, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (53/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,367] INFO [Log partition=__consumer_offsets-17, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:29,368] INFO Completed load of Log(dir=/tmp/kafka-logs/__consumer_offsets-17, topic=__consumer_offsets, partition=17, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=0) with 1 segments in 5ms (54/54 loaded in /tmp/kafka-logs) (kafka.log.LogManager)
[2021-06-04 01:20:29,370] INFO Loaded 54 logs in 746ms. (kafka.log.LogManager)
[2021-06-04 01:20:29,371] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
[2021-06-04 01:20:29,372] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
[2021-06-04 01:20:30,084] INFO Updated connection-accept-rate max connection creation rate to 2147483647 (kafka.network.ConnectionQuotas)
[2021-06-04 01:20:30,102] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.Acceptor)
[2021-06-04 01:20:30,164] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Created data-plane acceptor and processors for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-06-04 01:20:30,207] INFO [broker-0-to-controller-send-thread]: Starting (kafka.server.BrokerToControllerRequestThread)
[2021-06-04 01:20:30,255] INFO [ExpirationReaper-0-Produce]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,256] INFO [ExpirationReaper-0-Fetch]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,256] INFO [ExpirationReaper-0-DeleteRecords]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,257] INFO [ExpirationReaper-0-ElectLeader]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,286] INFO [LogDirFailureHandler]: Starting (kafka.server.ReplicaManager$LogDirFailureHandler)
[2021-06-04 01:20:30,388] INFO Creating /brokers/ids/0 (is it secure? false) (kafka.zk.KafkaZkClient)
[2021-06-04 01:20:30,423] INFO Stat of the created znode at /brokers/ids/0 is: 300,300,1622749830404,1622749830404,1,0,0,72059307042406401,202,0,300
 (kafka.zk.KafkaZkClient)
[2021-06-04 01:20:30,424] INFO Registered broker 0 at path /brokers/ids/0 with addresses: PLAINTEXT://localhost:9092, czxid (broker epoch): 300 (kafka.zk.KafkaZkClient)
[2021-06-04 01:20:30,583] INFO [ExpirationReaper-0-topic]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,596] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,598] INFO [ExpirationReaper-0-Rebalance]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,645] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:30,676] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:30,772] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:4000,blockEndProducerId:4999) by writing to Zk with path version 5 (kafka.coordinator.transaction.ProducerIdManager)
[2021-06-04 01:20:30,777] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2021-06-04 01:20:30,788] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2021-06-04 01:20:30,789] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2021-06-04 01:20:30,898] INFO [ExpirationReaper-0-AlterAcls]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-06-04 01:20:30,946] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2021-06-04 01:20:30,996] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Starting socket server acceptors and processors (kafka.network.SocketServer)
[2021-06-04 01:20:31,005] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-06-04 01:20:31,006] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started socket server acceptors and processors (kafka.network.SocketServer)
[2021-06-04 01:20:31,014] INFO Kafka version: 2.8.0 (org.apache.kafka.common.utils.AppInfoParser)
[2021-06-04 01:20:31,015] INFO Kafka commitId: ebb1d6e21cc92130 (org.apache.kafka.common.utils.AppInfoParser)
[2021-06-04 01:20:31,015] INFO Kafka startTimeMs: 1622749831006 (org.apache.kafka.common.utils.AppInfoParser)
[2021-06-04 01:20:31,017] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
[2021-06-04 01:20:31,146] 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-06-04 01:20:31,171] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions Set(__consumer_offsets-22, __consumer_offsets-30, __consumer_offsets-8, __consumer_offsets-21, __consumer_offsets-4, __consumer_offsets-27, __consumer_offsets-7, __consumer_offsets-9, __consumer_offsets-46, new_topic-0, NewTopic-0, __consumer_offsets-25, __consumer_offsets-35, __consumer_offsets-41, __consumer_offsets-33, __consumer_offsets-23, __consumer_offsets-49, __consumer_offsets-47, __consumer_offsets-16, __consumer_offsets-28, __consumer_offsets-31, __consumer_offsets-36, __consumer_offsets-42, __consumer_offsets-3, __consumer_offsets-18, __consumer_offsets-37, __consumer_offsets-15, __consumer_offsets-24, Messages-0, __consumer_offsets-38, __consumer_offsets-17, twitter_topic-0, __consumer_offsets-48, __consumer_offsets-19, __consumer_offsets-11, __consumer_offsets-13, __consumer_offsets-2, __consumer_offsets-43, __consumer_offsets-6, __consumer_offsets-14, __consumer_offsets-20, __consumer_offsets-0, __consumer_offsets-44, __consumer_offsets-39, __consumer_offsets-12, __consumer_offsets-45, messages-0, __consumer_offsets-1, __consumer_offsets-5, __consumer_offsets-26, __consumer_offsets-29, __consumer_offsets-34, __consumer_offsets-10, __consumer_offsets-32, __consumer_offsets-40) (kafka.server.ReplicaFetcherManager)
[2021-06-04 01:20:31,190] INFO [Partition __consumer_offsets-0 broker=0] Log loaded for partition __consumer_offsets-0 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,200] INFO [Partition __consumer_offsets-29 broker=0] Log loaded for partition __consumer_offsets-29 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,203] INFO [Partition __consumer_offsets-48 broker=0] Log loaded for partition __consumer_offsets-48 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,221] INFO [Log partition=Messages-0, dir=/tmp/kafka-logs] Loading producer state till offset 168 with message format version 2 (kafka.log.Log)
[2021-06-04 01:20:31,221] INFO [ProducerStateManager partition=Messages-0] Loading producer state from snapshot file 'SnapshotFile(/tmp/kafka-logs/Messages-0/00000000000000000168.snapshot,168)' (kafka.log.ProducerStateManager)
[2021-06-04 01:20:31,223] INFO Created log for partition Messages-0 in /tmp/kafka-logs/Messages-0 with properties {compression.type -> producer, message.downconversion.enable -> true, min.insync.replicas -> 1, segment.jitter.ms -> 0, cleanup.policy -> [delete], flush.ms -> 9223372036854775807, segment.bytes -> 1073741824, retention.ms -> 604800000, flush.messages -> 9223372036854775807, message.format.version -> 2.8-IV1, file.delete.delay.ms -> 60000, max.compaction.lag.ms -> 9223372036854775807, max.message.bytes -> 1048588, min.compaction.lag.ms -> 0, message.timestamp.type -> CreateTime, preallocate -> false, min.cleanable.dirty.ratio -> 0.5, index.interval.bytes -> 4096, unclean.leader.election.enable -> false, retention.bytes -> -1, delete.retention.ms -> 86400000, segment.ms -> 604800000, message.timestamp.difference.max.ms -> 9223372036854775807, segment.index.bytes -> 10485760}. (kafka.log.LogManager)
[2021-06-04 01:20:31,224] INFO [Partition Messages-0 broker=0] Log loaded for partition Messages-0 with initial high watermark 168 (kafka.cluster.Partition)
[2021-06-04 01:20:31,224] INFO [Partition __consumer_offsets-10 broker=0] Log loaded for partition __consumer_offsets-10 with initial high watermark 54 (kafka.cluster.Partition)
[2021-06-04 01:20:31,224] INFO [Partition __consumer_offsets-45 broker=0] Log loaded for partition __consumer_offsets-45 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,226] INFO [Partition __consumer_offsets-26 broker=0] Log loaded for partition __consumer_offsets-26 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,231] INFO [Partition __consumer_offsets-7 broker=0] Log loaded for partition __consumer_offsets-7 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,233] INFO [Partition __consumer_offsets-42 broker=0] Log loaded for partition __consumer_offsets-42 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,235] INFO [Partition __consumer_offsets-4 broker=0] Log loaded for partition __consumer_offsets-4 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,237] INFO [Partition __consumer_offsets-23 broker=0] Log loaded for partition __consumer_offsets-23 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,239] INFO [Partition NewTopic-0 broker=0] Log loaded for partition NewTopic-0 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,241] INFO [Partition __consumer_offsets-1 broker=0] Log loaded for partition __consumer_offsets-1 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,242] INFO [Partition __consumer_offsets-39 broker=0] Log loaded for partition __consumer_offsets-39 with initial high watermark 28 (kafka.cluster.Partition)
[2021-06-04 01:20:31,243] INFO [Partition __consumer_offsets-20 broker=0] Log loaded for partition __consumer_offsets-20 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,245] INFO [Partition __consumer_offsets-17 broker=0] Log loaded for partition __consumer_offsets-17 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,247] INFO [Partition __consumer_offsets-36 broker=0] Log loaded for partition __consumer_offsets-36 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,248] INFO [Partition __consumer_offsets-14 broker=0] Log loaded for partition __consumer_offsets-14 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,250] INFO [Partition __consumer_offsets-33 broker=0] Log loaded for partition __consumer_offsets-33 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,252] INFO [Partition __consumer_offsets-49 broker=0] Log loaded for partition __consumer_offsets-49 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,254] INFO [Partition __consumer_offsets-11 broker=0] Log loaded for partition __consumer_offsets-11 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,256] INFO [Partition __consumer_offsets-30 broker=0] Log loaded for partition __consumer_offsets-30 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,258] INFO [Partition __consumer_offsets-46 broker=0] Log loaded for partition __consumer_offsets-46 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,259] INFO [Partition __consumer_offsets-27 broker=0] Log loaded for partition __consumer_offsets-27 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,261] INFO [Partition __consumer_offsets-8 broker=0] Log loaded for partition __consumer_offsets-8 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,263] INFO [Partition __consumer_offsets-24 broker=0] Log loaded for partition __consumer_offsets-24 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,265] INFO [Partition __consumer_offsets-43 broker=0] Log loaded for partition __consumer_offsets-43 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,267] INFO [Partition __consumer_offsets-5 broker=0] Log loaded for partition __consumer_offsets-5 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,269] INFO [Partition messages-0 broker=0] Log loaded for partition messages-0 with initial high watermark 146 (kafka.cluster.Partition)
[2021-06-04 01:20:31,276] INFO [Partition __consumer_offsets-21 broker=0] Log loaded for partition __consumer_offsets-21 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,278] INFO [Partition __consumer_offsets-40 broker=0] Log loaded for partition __consumer_offsets-40 with initial high watermark 7 (kafka.cluster.Partition)
[2021-06-04 01:20:31,278] INFO [Partition __consumer_offsets-2 broker=0] Log loaded for partition __consumer_offsets-2 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,280] INFO [Partition __consumer_offsets-37 broker=0] Log loaded for partition __consumer_offsets-37 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,281] INFO [Partition __consumer_offsets-18 broker=0] Log loaded for partition __consumer_offsets-18 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,283] INFO [Partition new_topic-0 broker=0] Log loaded for partition new_topic-0 with initial high watermark 4 (kafka.cluster.Partition)
[2021-06-04 01:20:31,283] INFO [Partition __consumer_offsets-34 broker=0] Log loaded for partition __consumer_offsets-34 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,285] INFO [Partition __consumer_offsets-15 broker=0] Log loaded for partition __consumer_offsets-15 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,287] INFO [Partition twitter_topic-0 broker=0] Log loaded for partition twitter_topic-0 with initial high watermark 8 (kafka.cluster.Partition)
[2021-06-04 01:20:31,287] INFO [Partition __consumer_offsets-12 broker=0] Log loaded for partition __consumer_offsets-12 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,289] INFO [Partition __consumer_offsets-31 broker=0] Log loaded for partition __consumer_offsets-31 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,291] INFO [Partition __consumer_offsets-9 broker=0] Log loaded for partition __consumer_offsets-9 with initial high watermark 547 (kafka.cluster.Partition)
[2021-06-04 01:20:31,291] INFO [Partition __consumer_offsets-47 broker=0] Log loaded for partition __consumer_offsets-47 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,293] INFO [Partition __consumer_offsets-19 broker=0] Log loaded for partition __consumer_offsets-19 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,294] INFO [Partition __consumer_offsets-28 broker=0] Log loaded for partition __consumer_offsets-28 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,296] INFO [Partition __consumer_offsets-38 broker=0] Log loaded for partition __consumer_offsets-38 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,298] INFO [Partition __consumer_offsets-35 broker=0] Log loaded for partition __consumer_offsets-35 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,300] INFO [Partition __consumer_offsets-6 broker=0] Log loaded for partition __consumer_offsets-6 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,302] INFO [Partition __consumer_offsets-44 broker=0] Log loaded for partition __consumer_offsets-44 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,303] INFO [Partition __consumer_offsets-25 broker=0] Log loaded for partition __consumer_offsets-25 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,304] INFO [Partition __consumer_offsets-16 broker=0] Log loaded for partition __consumer_offsets-16 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,306] INFO [Partition __consumer_offsets-22 broker=0] Log loaded for partition __consumer_offsets-22 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,309] INFO [Partition __consumer_offsets-41 broker=0] Log loaded for partition __consumer_offsets-41 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,311] INFO [Partition __consumer_offsets-32 broker=0] Log loaded for partition __consumer_offsets-32 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,313] INFO [Partition __consumer_offsets-3 broker=0] Log loaded for partition __consumer_offsets-3 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,314] INFO [Partition __consumer_offsets-13 broker=0] Log loaded for partition __consumer_offsets-13 with initial high watermark 0 (kafka.cluster.Partition)
[2021-06-04 01:20:31,328] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 22 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,329] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-22 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,331] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 25 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,331] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-25 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,331] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 28 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,331] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-28 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 31 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-31 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 34 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-34 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 37 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-37 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 40 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-40 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 43 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-43 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 46 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-46 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 49 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,332] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-49 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,332] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 41 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-41 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 44 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-44 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 47 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-47 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 1 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-1 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 4 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-4 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 7 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-7 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 10 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-10 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,333] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 16 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,333] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-16 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 19 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-19 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 13 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-13 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 2 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-2 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 5 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-5 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 8 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-8 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 11 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-11 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 14 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-14 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 17 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-17 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 20 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-20 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 23 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-23 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 26 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-26 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 29 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,334] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-29 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,334] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 32 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,335] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-32 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,335] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 35 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,335] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-35 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,335] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 38 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,335] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-38 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,335] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 0 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,335] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-0 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,335] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 3 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-3 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,336] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 6 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-6 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,336] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 9 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-9 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,336] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 15 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-15 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,336] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 18 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-18 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,336] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 21 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-21 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,336] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 24 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,336] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-24 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 27 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,337] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-27 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 30 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,337] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-30 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 33 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,337] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-33 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 36 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,337] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-36 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 39 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,337] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-39 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 42 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,337] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-42 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,337] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 45 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,338] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-45 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,338] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 48 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,338] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-48 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,338] INFO [GroupCoordinator 0]: Elected as the group coordinator for partition 12 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,338] INFO [GroupMetadataManager brokerId=0] Scheduling loading of offsets and group metadata from __consumer_offsets-12 (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,340] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-22 in 10 milliseconds, of which 1 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,341] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-25 in 10 milliseconds, of which 10 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,342] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-28 in 11 milliseconds, of which 10 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,342] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-31 in 10 milliseconds, of which 10 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,342] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-34 in 10 milliseconds, of which 10 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,342] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-37 in 10 milliseconds, of which 10 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,417] INFO Loaded member MemberMetadata(memberId=consumer-group1-1-2b58f9d9-fac5-4f53-bb58-ce85fe2e5c08, groupInstanceId=None, clientId=consumer-group1-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group group1 with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,435] INFO [GroupCoordinator 0]: Loading group metadata for group1 with generation 2 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,436] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-40 in 104 milliseconds, of which 10 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,437] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-43 in 105 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,437] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-46 in 105 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,437] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-49 in 105 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,438] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-41 in 105 milliseconds, of which 104 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,438] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-44 in 105 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,438] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-47 in 105 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,438] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-1 in 105 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,439] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-4 in 106 milliseconds, of which 105 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,439] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-7 in 106 milliseconds, of which 106 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,446] INFO Loaded member MemberMetadata(memberId=consumer-1-0a13bce6-e504-408e-99db-23fab25e9286, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,447] INFO Loaded member MemberMetadata(memberId=consumer-1-f2bd9d55-d5a6-45bd-971f-4a1269c2b062, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,449] INFO Loaded member MemberMetadata(memberId=consumer-1-31d6f448-0201-401c-b8b3-a2dd43c7c1ce, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,449] INFO Loaded member MemberMetadata(memberId=consumer-1-aa03b97a-dc31-4ef6-9b5c-26f1a6df6223, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,450] INFO Loaded member MemberMetadata(memberId=consumer-1-deee6208-9549-41a1-a4b9-8f3a1e4c16fd, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,451] INFO Loaded member MemberMetadata(memberId=consumer-1-5f9ef16a-36d4-4781-bf24-cff13d42b271, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,451] INFO Loaded member MemberMetadata(memberId=consumer-1-a8f985c6-61a8-4bcd-942c-fdf9e0f5385e, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,452] INFO Loaded member MemberMetadata(memberId=consumer-1-145e1be2-fbd6-40cb-9ea6-0afed81d91e6, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 5. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,452] INFO Loaded member MemberMetadata(memberId=consumer-1-e587653f-65f5-400e-bd37-1fd0c1946fdc, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,453] INFO Loaded member MemberMetadata(memberId=consumer-1-d11640e0-deee-4982-8d64-b05b8ab22c5c, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,454] INFO Loaded member MemberMetadata(memberId=consumer-1-e9e152f0-b73d-445c-809a-084022a9aa99, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,454] INFO Loaded member MemberMetadata(memberId=consumer-1-4bed9ac4-ad40-41bd-b551-5c527beef6e8, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,455] INFO Loaded member MemberMetadata(memberId=consumer-1-d4cacaa4-aba0-4ab7-8cc9-15157cc0fb21, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,455] INFO Loaded member MemberMetadata(memberId=consumer-1-293c33e5-4c4a-4e6e-95f6-412db3d11e41, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,456] INFO Loaded member MemberMetadata(memberId=consumer-1-54bbebf5-23c5-42d9-8662-6348c7e283e4, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,456] INFO Loaded member MemberMetadata(memberId=consumer-1-0953a32a-1489-40a9-ba55-cc35d8324b81, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,457] INFO Loaded member MemberMetadata(memberId=consumer-1-267970a4-320d-4159-8f47-aaccbbfbee1a, groupInstanceId=None, clientId=consumer-1, clientHost=/127.0.0.1, sessionTimeoutMs=30000, rebalanceTimeoutMs=30000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,457] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream-1-e8cfe033-42df-4fe7-858e-5033e2a92155, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,457] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream-1-b15ee680-e159-43a5-b176-468de4b10ae3, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,458] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream-1-a8daaeb0-9913-427f-8846-a72370180bd7, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 5. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,459] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream-1-9b188a70-0619-4d56-8a38-10eccf6b877a, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,459] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-10 in 126 milliseconds, of which 106 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,460] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-16 in 126 milliseconds, of which 125 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,460] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-19 in 126 milliseconds, of which 126 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,460] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-13 in 126 milliseconds, of which 126 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,460] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-2 in 126 milliseconds, of which 126 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,460] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-5 in 126 milliseconds, of which 126 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,461] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-8 in 127 milliseconds, of which 127 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,461] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-11 in 127 milliseconds, of which 127 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,461] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-14 in 127 milliseconds, of which 127 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,461] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-17 in 127 milliseconds, of which 127 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,462] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-20 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,462] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-23 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,462] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-26 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,463] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-29 in 129 milliseconds, of which 129 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,463] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-32 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,463] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-35 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,463] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-38 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,464] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-0 in 129 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,464] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-3 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,464] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-6 in 128 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,471] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream_-1-3c29bd84-084b-4854-ae27-e1ed70547acc, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream_-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream_ with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,486] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream_-1-8b1c2ec4-de0b-47db-9604-52cb9cc7bb3d, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream_-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream_ with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,487] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream_-1-e89a69d7-a02f-44ce-bd32-6e6e0e2ac533, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream_-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream_ with generation 5. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,491] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream_-1-a78a6fb8-d01d-4deb-ab8d-892566b3bbf0, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream_-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream_ with generation 7. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,492] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream_-1-4171f275-03bd-43ac-8c7e-46b30ef709ee, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream_-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream_ with generation 9. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,495] INFO Loaded member MemberMetadata(memberId=consumer-use_a_separate_group_id_for_each_stream_-1-11b034fa-4448-44ef-92ef-264941fee15f, groupInstanceId=None, clientId=consumer-use_a_separate_group_id_for_each_stream_-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group use_a_separate_group_id_for_each_stream_ with generation 11. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,499] INFO [GroupCoordinator 0]: Loading group metadata for use_a_separate_group_id_for_each_stream_ with generation 12 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,499] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-9 in 163 milliseconds, of which 128 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,500] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-15 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,500] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-18 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,500] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-21 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,501] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-24 in 165 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,501] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-27 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,501] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-30 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,501] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-33 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,501] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-36 in 164 milliseconds, of which 164 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,511] INFO Loaded member MemberMetadata(memberId=consumer-group2-1-de0bb6b1-ee25-4b14-8135-3259618ceef5, groupInstanceId=None, clientId=consumer-group2-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group group2 with generation 1. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,512] INFO Loaded member MemberMetadata(memberId=consumer-group2-1-5cb9542c-6c0a-40ed-9f76-a4cf97f27921, groupInstanceId=None, clientId=consumer-group2-1, clientHost=/127.0.0.1, sessionTimeoutMs=10000, rebalanceTimeoutMs=300000, supportedProtocols=List(range), ) in group group2 with generation 3. (kafka.coordinator.group.GroupMetadata$)
[2021-06-04 01:20:31,513] INFO [GroupCoordinator 0]: Loading group metadata for group2 with generation 4 (kafka.coordinator.group.GroupCoordinator)
[2021-06-04 01:20:31,513] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-39 in 176 milliseconds, of which 165 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,513] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-42 in 176 milliseconds, of which 176 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,513] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-45 in 175 milliseconds, of which 175 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,514] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-48 in 176 milliseconds, of which 175 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:31,514] INFO [GroupMetadataManager brokerId=0] Finished loading offsets and group metadata from __consumer_offsets-12 in 176 milliseconds, of which 176 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
[2021-06-04 01:20:59,468] INFO [Log partition=Messages-0, dir=/tmp/kafka-logs] Rolled new log segment at offset 168 in 22 ms. (kafka.log.Log)
[2021-06-04 01:20:59,471] INFO [Log partition=Messages-0, dir=/tmp/kafka-logs] Deleting segment LogSegment(baseOffset=146, size=1810, lastModifiedTime=1622740000000, largestRecordTimestamp=Some(0)) due to retention time 604800000ms breach based on the largest record timestamp in the segment (kafka.log.Log)
[2021-06-04 01:20:59,480] INFO [Log partition=Messages-0, dir=/tmp/kafka-logs] Incremented log start offset to 168 due to segment deletion (kafka.log.Log)

2.5 Spark Integration Example

Apache Spark is based on Hadoop and its key components HDFS and YARN. Spark release consists of Hadoop packages. Let us look at an example that shows the Spark and Kafka integration. Topics are configured on Kafka. Spark connects to Kafka’s topic to receive a stream of messages. Messages can be processed in real-time. The source code is shown below for the example.

Apache KafkaSpark Integration Example

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.TaskContext;
import org.apache.spark.streaming.Durations;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;
import java.util.regex.Pattern;

import javafx.util.Duration;



public class KafkaSparkIntegrationExample {
	

	
	public static void main(String[] arr)
	{
		SparkConf sparkConf = new SparkConf();
		sparkConf.setAppName("kafkastreamingclient");
		sparkConf.setMaster("local[*]");

		
		JavaStreamingContext kafkaStreamingContext = new JavaStreamingContext(sparkConf,Durations.seconds(1));
		Collection topics = Arrays.asList("new_topic");
		
		Map kafkaParameters =  new HashMap();
		kafkaParameters.put("bootstrap.servers", "localhost:9092");
		kafkaParameters.put("value.deserializer", StringDeserializer.class);
		kafkaParameters.put("key.deserializer", StringDeserializer.class);
		kafkaParameters.put("group.id", "use_a_separate_group_id_for_each_stream_");
		kafkaParameters.put("auto.offset.reset", "latest");
		kafkaParameters.put("enable.auto.commit", false);
		JavaInputDStream<ConsumerRecord> kafkaStream = KafkaUtils.createDirectStream(kafkaStreamingContext, LocationStrategies.PreferConsistent(),
				ConsumerStrategies.Subscribe(topics, kafkaParameters)
				  );
		
	
		JavaDStream message = kafkaStream.map(entry->entry.value());

		message.foreachRDD(kafkardd -> { 
									System.out.println("Count of messages:"+kafkardd.count());  
									kafkardd.foreach(value->System.out.println(value)); }
			  
							);

		kafkaStream.foreachRDD(rdd -> {
			  OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
			  rdd.foreachPartition(consumerRecords -> {
			    OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
			    
			   });
			  ((CanCommitOffsets) kafkaStream.inputDStream()).commitAsync(offsetRanges);

			});
		
	        
		kafkaStreamingContext.start();
		try {
			kafkaStreamingContext.awaitTermination();
		} catch (InterruptedException exception) {

			exception.printStackTrace();
		}
		
	}

}

The command to compile the above code is shown below:

Compilation Command

mvn package

The output of the above command is shown as below:

Compilation Output

apples-MacBook-Air:apachekafkaspark bhagvan.kommadi$ mvn package
[INFO] Scanning for projects...
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for javacodegeeks:apache-kafka-spark:jar:1.0
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-compiler-plugin is missing. @ line 67, column 16
[WARNING] 
[WARNING] It is highly recommended to fix these problems because they threaten the stability of your build.
[WARNING] 
[WARNING] For this reason, future Maven versions might no longer support building such malformed projects.
[WARNING] 
[INFO] 
[INFO] ------------------------------------
[INFO] Building apache-kafka-spark 1.0
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- maven-resources-plugin:2.6:resources (default-resources) @ apache-kafka-spark ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] skip non existing resourceDirectory /Users/bhagvan.kommadi/Desktop/JavacodeGeeks/Code/apachekafkaspark/src/main/resources
[INFO] 
[INFO] --- maven-compiler-plugin:3.1:compile (default-compile) @ apache-kafka-spark ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 1 source file to /Users/bhagvan.kommadi/Desktop/JavacodeGeeks/Code/apachekafkaspark/target/classes
[INFO] 
[INFO] --- maven-resources-plugin:2.6:testResources (default-testResources) @ apache-kafka-spark ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] skip non existing resourceDirectory /Users/bhagvan.kommadi/Desktop/JavacodeGeeks/Code/apachekafkaspark/src/test/resources
[INFO] 
[INFO] --- maven-compiler-plugin:3.1:testCompile (default-testCompile) @ apache-kafka-spark ---
[INFO] Nothing to compile - all classes are up to date
[INFO] 
[INFO] --- maven-surefire-plugin:2.12.4:test (default-test) @ apache-kafka-spark ---
[INFO] No tests to run.
[INFO] 
[INFO] --- maven-jar-plugin:2.4:jar (default-jar) @ apache-kafka-spark ---
[INFO] Building jar: /Users/bhagvan.kommadi/Desktop/JavacodeGeeks/Code/apachekafkaspark/target/apache-kafka-spark-1.0.jar
[INFO] 
[INFO] --- maven-assembly-plugin:3.3.0:single (default) @ apache-kafka-spark ---
[INFO] Building jar: /Users/bhagvan.kommadi/Desktop/JavacodeGeeks/Code/apachekafkaspark/target/apache-kafka-spark-1.0-jar-with-dependencies.jar
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  53.160 s
[INFO] Finished at: 2021-06-04T00:39:42+05:30
[INFO] ------------------------------------------------------------------------
apples-MacBook-Air:apachekafkaspark bhagvan.kommadi$

The command for executing the example is shown below:

Execution Command

 mvn exec:java -Dexec.mainClass=KafkaSparkIntegrationExample 

You can create a topic by using this command :

Kafka producer CLI

./bin/kafka-topics.sh --create   --zookeeper localhost:2181   --replication-factor 1 --partitions 1   --topic new_topic

The output of the above CLI command is shown as below:

Kafka producer CLI Command Output

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$ ./bin/kafka-topics.sh --create   --zookeeper localhost:2181   --replication-factor 1 --partitions 1   --topic new_topic
Created topic new_topic.

You can send messages by using kafka console producer CLI as shown below:

Kafka producer CLI

./bin/kafka-console-producer.sh     --broker-list localhost:9092     --topic new_topic

The output of the above CLI command is shown as below:

Kafka producer CLI Output

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$ ./bin/kafka-console-producer.sh     --broker-list localhost:9092     --topic new_topic
>Kafka topic spark factor
>zookeeper server message
>this is a test
>scalable architecture

apples-MacBook-Air:kafka_2.12-2.8.0 bhagvan.kommadi$

The output of the above example code executed is shown as below:

Apache Kafka Producer Execution Output

apples-MacBook-Air:apachekafkaspark bhagvan.kommadi$ mvn exec:java -Dexec.mainClass=KafkaSparkIntegrationExample
[INFO] Scanning for projects...
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for javacodegeeks:apache-kafka-spark:jar:1.0
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-compiler-plugin is missing. @ line 67, column 16
[WARNING] 
[WARNING] It is highly recommended to fix these problems because they threaten the stability of your build.
[WARNING] 
[WARNING] For this reason, future Maven versions might no longer support building such malformed projects.
[WARNING] 
[INFO] 
[INFO] ------------------------------------
[INFO] Building apache-kafka-spark 1.0
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- exec-maven-plugin:3.0.0:java (default-cli) @ apache-kafka-spark ---
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
21/06/04 00:41:49 WARN Utils: Your hostname, apples-MacBook-Air.local resolves to a loopback address: 127.0.0.1; using 192.168.1.7 instead (on interface en0)
21/06/04 00:41:49 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
21/06/04 00:42:20 INFO SparkContext: Running Spark version 3.1.2
21/06/04 00:42:20 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
21/06/04 00:42:20 INFO ResourceUtils: ==============================================================
21/06/04 00:42:20 INFO ResourceUtils: No custom resources configured for spark.driver.
21/06/04 00:42:20 INFO ResourceUtils: ==============================================================
21/06/04 00:42:20 INFO SparkContext: Submitted application: kafkastreamingclient
21/06/04 00:42:20 INFO ResourceProfile: Default ResourceProfile created, executor resources: Map(cores -> name: cores, amount: 1, script: , vendor: , memory -> name: memory, amount: 1024, script: , vendor: , offHeap -> name: offHeap, amount: 0, script: , vendor: ), task resources: Map(cpus -> name: cpus, amount: 1.0)
21/06/04 00:42:20 INFO ResourceProfile: Limiting resource is cpu
21/06/04 00:42:20 INFO ResourceProfileManager: Added ResourceProfile id: 0
21/06/04 00:42:21 INFO SecurityManager: Changing view acls to: bhagvan.kommadi
21/06/04 00:42:21 INFO SecurityManager: Changing modify acls to: bhagvan.kommadi
21/06/04 00:42:21 INFO SecurityManager: Changing view acls groups to: 
21/06/04 00:42:21 INFO SecurityManager: Changing modify acls groups to: 
21/06/04 00:42:21 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(bhagvan.kommadi); groups with view permissions: Set(); users  with modify permissions: Set(bhagvan.kommadi); groups with modify permissions: Set()
21/06/04 00:42:22 INFO Utils: Successfully started service 'sparkDriver' on port 52758.
21/06/04 00:42:22 INFO SparkEnv: Registering MapOutputTracker
21/06/04 00:42:22 INFO SparkEnv: Registering BlockManagerMaster
21/06/04 00:42:22 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
21/06/04 00:42:22 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
21/06/04 00:42:22 INFO SparkEnv: Registering BlockManagerMasterHeartbeat
21/06/04 00:42:22 INFO DiskBlockManager: Created local directory at /private/var/folders/cr/0y892lq14qv7r24yl0gh0_dm0000gp/T/blockmgr-2245ef95-07b7-46ab-9a2e-45f996abebc5
21/06/04 00:42:22 INFO MemoryStore: MemoryStore started with capacity 366.3 MiB
21/06/04 00:42:22 INFO SparkEnv: Registering OutputCommitCoordinator
21/06/04 00:42:23 INFO Utils: Successfully started service 'SparkUI' on port 4040.
21/06/04 00:42:23 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://192.168.1.7:4040
21/06/04 00:42:23 INFO Executor: Starting executor ID driver on host 192.168.1.7
21/06/04 00:42:23 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 52779.
21/06/04 00:42:23 INFO NettyBlockTransferService: Server created on 192.168.1.7:52779
21/06/04 00:42:23 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
21/06/04 00:42:23 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.1.7, 52779, None)
21/06/04 00:42:23 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.1.7:52779 with 366.3 MiB RAM, BlockManagerId(driver, 192.168.1.7, 52779, None)
21/06/04 00:42:23 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168.1.7, 52779, None)
21/06/04 00:42:23 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168.1.7, 52779, None)
21/06/04 00:42:24 WARN KafkaUtils: overriding enable.auto.commit to false for executor
21/06/04 00:42:24 WARN KafkaUtils: overriding auto.offset.reset to none for executor
21/06/04 00:42:24 WARN KafkaUtils: overriding executor group.id to spark-executor-use_a_separate_group_id_for_each_stream_
21/06/04 00:42:24 WARN KafkaUtils: overriding receive.buffer.bytes to 65536 see KAFKA-3135
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Slide time = 1000 ms
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Storage level = Serialized 1x Replicated
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Checkpoint interval = null
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Remember interval = 1000 ms
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Initialized and validated org.apache.spark.streaming.kafka010.DirectKafkaInputDStream@4c0c95fb
21/06/04 00:42:24 INFO MappedDStream: Slide time = 1000 ms
21/06/04 00:42:24 INFO MappedDStream: Storage level = Serialized 1x Replicated
21/06/04 00:42:24 INFO MappedDStream: Checkpoint interval = null
21/06/04 00:42:24 INFO MappedDStream: Remember interval = 1000 ms
21/06/04 00:42:24 INFO MappedDStream: Initialized and validated org.apache.spark.streaming.dstream.MappedDStream@490a4ad2
21/06/04 00:42:24 INFO ForEachDStream: Slide time = 1000 ms
21/06/04 00:42:24 INFO ForEachDStream: Storage level = Serialized 1x Replicated
21/06/04 00:42:24 INFO ForEachDStream: Checkpoint interval = null
21/06/04 00:42:24 INFO ForEachDStream: Remember interval = 1000 ms
21/06/04 00:42:24 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream@14d162fc
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Slide time = 1000 ms
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Storage level = Serialized 1x Replicated
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Checkpoint interval = null
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Remember interval = 1000 ms
21/06/04 00:42:24 INFO DirectKafkaInputDStream: Initialized and validated org.apache.spark.streaming.kafka010.DirectKafkaInputDStream@4c0c95fb
21/06/04 00:42:24 INFO ForEachDStream: Slide time = 1000 ms
21/06/04 00:42:24 INFO ForEachDStream: Storage level = Serialized 1x Replicated
21/06/04 00:42:24 INFO ForEachDStream: Checkpoint interval = null
21/06/04 00:42:24 INFO ForEachDStream: Remember interval = 1000 ms
21/06/04 00:42:24 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream@404a2b1d
21/06/04 00:42:25 INFO ConsumerConfig: ConsumerConfig values: 
	allow.auto.create.topics = true
	auto.commit.interval.ms = 5000
	auto.offset.reset = latest
	bootstrap.servers = [localhost:9092]
	check.crcs = true
	client.dns.lookup = use_all_dns_ips
	client.id = consumer-use_a_separate_group_id_for_each_stream_-1
	client.rack = 
	connections.max.idle.ms = 540000
	default.api.timeout.ms = 60000
	enable.auto.commit = false
	exclude.internal.topics = true
	fetch.max.bytes = 52428800
	fetch.max.wait.ms = 500
	fetch.min.bytes = 1
	group.id = use_a_separate_group_id_for_each_stream_
	group.instance.id = null
	heartbeat.interval.ms = 3000
	interceptor.classes = []
	internal.leave.group.on.close = true
	internal.throw.on.fetch.stable.offset.unsupported = false
	isolation.level = read_uncommitted
	key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
	max.partition.fetch.bytes = 1048576
	max.poll.interval.ms = 300000
	max.poll.records = 500
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	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 = GSSAPI
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	ssl.cipher.suites = null
	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.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

21/06/04 00:42:25 INFO AppInfoParser: Kafka version: 2.6.0
21/06/04 00:42:25 INFO AppInfoParser: Kafka commitId: 62abe01bee039651
21/06/04 00:42:25 INFO AppInfoParser: Kafka startTimeMs: 1622747545420
21/06/04 00:42:25 INFO KafkaConsumer: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Subscribed to topic(s): new_topic
21/06/04 00:42:26 INFO Metadata: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Cluster ID: s6HNO0aXTMC5pyby0hV2kw
21/06/04 00:42:26 INFO AbstractCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Discovered group coordinator localhost:9092 (id: 2147483647 rack: null)
21/06/04 00:42:26 INFO AbstractCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] (Re-)joining group
21/06/04 00:42:26 INFO AbstractCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Join group failed with org.apache.kafka.common.errors.MemberIdRequiredException: The group member needs to have a valid member id before actually entering a consumer group.
21/06/04 00:42:26 INFO AbstractCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] (Re-)joining group
21/06/04 00:42:26 INFO ConsumerCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Finished assignment for group at generation 11: {consumer-use_a_separate_group_id_for_each_stream_-1-11b034fa-4448-44ef-92ef-264941fee15f=Assignment(partitions=[new_topic-0])}
21/06/04 00:42:26 INFO AbstractCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Successfully joined group with generation 11
21/06/04 00:42:26 INFO ConsumerCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Notifying assignor about the new Assignment(partitions=[new_topic-0])
21/06/04 00:42:26 INFO ConsumerCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Adding newly assigned partitions: new_topic-0
21/06/04 00:42:26 INFO ConsumerCoordinator: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Found no committed offset for partition new_topic-0
21/06/04 00:42:26 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Resetting offset for partition new_topic-0 to offset 0.
21/06/04 00:42:26 INFO RecurringTimer: Started timer for JobGenerator at time 1622747545000
21/06/04 00:42:26 INFO JobGenerator: Started JobGenerator at 1622747545000 ms
21/06/04 00:42:26 INFO JobScheduler: Started JobScheduler
21/06/04 00:42:26 INFO StreamingContext: StreamingContext started
21/06/04 00:42:26 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Seeking to LATEST offset of partition new_topic-0
21/06/04 00:42:26 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Resetting offset for partition new_topic-0 to offset 0.
21/06/04 00:42:26 INFO JobScheduler: Added jobs for time 1622747545000 ms
21/06/04 00:42:26 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Seeking to LATEST offset of partition new_topic-0
21/06/04 00:42:26 INFO JobScheduler: Starting job streaming job 1622747545000 ms.0 from job set of time 1622747545000 ms
21/06/04 00:42:26 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Resetting offset for partition new_topic-0 to offset 0.
21/06/04 00:42:26 INFO JobScheduler: Added jobs for time 1622747546000 ms
21/06/04 00:42:26 INFO SparkContext: Starting job: count at KafkaSparkIntegrationExample.java:49
21/06/04 00:42:26 INFO DAGScheduler: Got job 0 (count at KafkaSparkIntegrationExample.java:49) with 1 output partitions
21/06/04 00:42:26 INFO DAGScheduler: Final stage: ResultStage 0 (count at KafkaSparkIntegrationExample.java:49)
21/06/04 00:42:26 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:26 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:26 INFO DAGScheduler: Submitting ResultStage 0 (MapPartitionsRDD[1] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:27 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Seeking to LATEST offset of partition new_topic-0
21/06/04 00:42:27 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Resetting offset for partition new_topic-0 to offset 0.
21/06/04 00:42:27 INFO JobScheduler: Added jobs for time 1622747547000 ms
21/06/04 00:42:27 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 4.4 KiB, free 366.3 MiB)
21/06/04 00:42:27 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 2.7 KiB, free 366.3 MiB)
21/06/04 00:42:27 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on 192.168.1.7:52779 (size: 2.7 KiB, free: 366.3 MiB)
21/06/04 00:42:27 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:27 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 0 (MapPartitionsRDD[1] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:27 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks resource profile 0
21/06/04 00:42:27 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:27 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
21/06/04 00:42:28 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Seeking to LATEST offset of partition new_topic-0
21/06/04 00:42:28 INFO SubscriptionState: [Consumer clientId=consumer-use_a_separate_group_id_for_each_stream_-1, groupId=use_a_separate_group_id_for_each_stream_] Resetting offset for partition new_topic-0 to offset 0.
21/06/04 00:42:28 INFO JobScheduler: Added jobs for time 1622747548000 ms
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 918 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 781 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 0 (count at KafkaSparkIntegrationExample.java:49) finished in 1.140 s
21/06/04 00:42:28 INFO DAGScheduler: Job 0 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 0: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 0 finished: count at KafkaSparkIntegrationExample.java:49, took 1.252545 s
Count of messages:0
21/06/04 00:42:28 INFO SparkContext: Starting job: foreach at KafkaSparkIntegrationExample.java:50
21/06/04 00:42:28 INFO DAGScheduler: Got job 1 (foreach at KafkaSparkIntegrationExample.java:50) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 1 (foreach at KafkaSparkIntegrationExample.java:50)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 1 (MapPartitionsRDD[1] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 4.9 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 2.8 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on 192.168.1.7:52779 (size: 2.8 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[1] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 1.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 1) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 1.0 (TID 1)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 1.0 (TID 1). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 1.0 (TID 1) in 53 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 1 (foreach at KafkaSparkIntegrationExample.java:50) finished in 0.075 s
21/06/04 00:42:28 INFO DAGScheduler: Job 1 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 1: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 1 finished: foreach at KafkaSparkIntegrationExample.java:50, took 0.082930 s
21/06/04 00:42:28 INFO JobScheduler: Finished job streaming job 1622747545000 ms.0 from job set of time 1622747545000 ms
21/06/04 00:42:28 INFO JobScheduler: Starting job streaming job 1622747545000 ms.1 from job set of time 1622747545000 ms
21/06/04 00:42:28 INFO SparkContext: Starting job: foreachPartition at KafkaSparkIntegrationExample.java:56
21/06/04 00:42:28 INFO DAGScheduler: Got job 2 (foreachPartition at KafkaSparkIntegrationExample.java:56) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 2 (foreachPartition at KafkaSparkIntegrationExample.java:56)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 2 (KafkaRDD[0] at createDirectStream at KafkaSparkIntegrationExample.java:41), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 3.5 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 2.1 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_2_piece0 in memory on 192.168.1.7:52779 (size: 2.1 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 2 (KafkaRDD[0] at createDirectStream at KafkaSparkIntegrationExample.java:41) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 2.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 2.0 (TID 2) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 2.0 (TID 2)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 2.0 (TID 2). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 2.0 (TID 2) in 22 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 2 (foreachPartition at KafkaSparkIntegrationExample.java:56) finished in 0.059 s
21/06/04 00:42:28 INFO DAGScheduler: Job 2 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 2: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 2 finished: foreachPartition at KafkaSparkIntegrationExample.java:56, took 0.074635 s
21/06/04 00:42:28 INFO JobScheduler: Finished job streaming job 1622747545000 ms.1 from job set of time 1622747545000 ms
21/06/04 00:42:28 INFO JobScheduler: Total delay: 3.367 s for time 1622747545000 ms (execution: 1.599 s)
21/06/04 00:42:28 INFO JobScheduler: Starting job streaming job 1622747546000 ms.0 from job set of time 1622747546000 ms
21/06/04 00:42:28 INFO ReceivedBlockTracker: Deleting batches: 
21/06/04 00:42:28 INFO SparkContext: Starting job: count at KafkaSparkIntegrationExample.java:49
21/06/04 00:42:28 INFO DAGScheduler: Got job 3 (count at KafkaSparkIntegrationExample.java:49) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 3 (count at KafkaSparkIntegrationExample.java:49)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 3 (MapPartitionsRDD[3] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO InputInfoTracker: remove old batch metadata: 
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 4.4 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 2.7 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_3_piece0 in memory on 192.168.1.7:52779 (size: 2.7 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 3 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 3 (MapPartitionsRDD[3] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 3.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 3.0 (TID 3) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 3.0 (TID 3)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 3.0 (TID 3). 875 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 3.0 (TID 3) in 13 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 3.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 3 (count at KafkaSparkIntegrationExample.java:49) finished in 0.035 s
21/06/04 00:42:28 INFO DAGScheduler: Job 3 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 3: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 3 finished: count at KafkaSparkIntegrationExample.java:49, took 0.049286 s
Count of messages:0
21/06/04 00:42:28 INFO SparkContext: Starting job: foreach at KafkaSparkIntegrationExample.java:50
21/06/04 00:42:28 INFO DAGScheduler: Got job 4 (foreach at KafkaSparkIntegrationExample.java:50) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 4 (foreach at KafkaSparkIntegrationExample.java:50)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 4 (MapPartitionsRDD[3] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_4 stored as values in memory (estimated size 4.9 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_4_piece0 stored as bytes in memory (estimated size 2.8 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_4_piece0 in memory on 192.168.1.7:52779 (size: 2.8 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 4 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 4 (MapPartitionsRDD[3] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 4.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 4.0 (TID 4) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 4.0 (TID 4)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 4.0 (TID 4). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 4.0 (TID 4) in 11 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 4.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 4 (foreach at KafkaSparkIntegrationExample.java:50) finished in 0.064 s
21/06/04 00:42:28 INFO DAGScheduler: Job 4 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 4: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 4 finished: foreach at KafkaSparkIntegrationExample.java:50, took 0.075511 s
21/06/04 00:42:28 INFO JobScheduler: Finished job streaming job 1622747546000 ms.0 from job set of time 1622747546000 ms
21/06/04 00:42:28 INFO JobScheduler: Starting job streaming job 1622747546000 ms.1 from job set of time 1622747546000 ms
21/06/04 00:42:28 INFO BlockManagerInfo: Removed broadcast_0_piece0 on 192.168.1.7:52779 in memory (size: 2.7 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Removed broadcast_1_piece0 on 192.168.1.7:52779 in memory (size: 2.8 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Starting job: foreachPartition at KafkaSparkIntegrationExample.java:56
21/06/04 00:42:28 INFO DAGScheduler: Got job 5 (foreachPartition at KafkaSparkIntegrationExample.java:56) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 5 (foreachPartition at KafkaSparkIntegrationExample.java:56)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO BlockManagerInfo: Removed broadcast_2_piece0 on 192.168.1.7:52779 in memory (size: 2.1 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 5 (KafkaRDD[2] at createDirectStream at KafkaSparkIntegrationExample.java:41), which has no missing parents
21/06/04 00:42:28 INFO BlockManagerInfo: Removed broadcast_3_piece0 on 192.168.1.7:52779 in memory (size: 2.7 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_5 stored as values in memory (estimated size 3.5 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_5_piece0 stored as bytes in memory (estimated size 2.1 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_5_piece0 in memory on 192.168.1.7:52779 (size: 2.1 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 5 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 5 (KafkaRDD[2] at createDirectStream at KafkaSparkIntegrationExample.java:41) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 5.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 5.0 (TID 5) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 5.0 (TID 5)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 5.0 (TID 5). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 5.0 (TID 5) in 9 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 5.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 5 (foreachPartition at KafkaSparkIntegrationExample.java:56) finished in 0.026 s
21/06/04 00:42:28 INFO DAGScheduler: Job 5 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 5: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 5 finished: foreachPartition at KafkaSparkIntegrationExample.java:56, took 0.042010 s
21/06/04 00:42:28 INFO JobScheduler: Finished job streaming job 1622747546000 ms.1 from job set of time 1622747546000 ms
21/06/04 00:42:28 INFO JobScheduler: Total delay: 2.614 s for time 1622747546000 ms (execution: 0.241 s)
21/06/04 00:42:28 INFO MapPartitionsRDD: Removing RDD 1 from persistence list
21/06/04 00:42:28 INFO JobScheduler: Starting job streaming job 1622747547000 ms.0 from job set of time 1622747547000 ms
21/06/04 00:42:28 INFO BlockManager: Removing RDD 1
21/06/04 00:42:28 INFO KafkaRDD: Removing RDD 0 from persistence list
21/06/04 00:42:28 INFO BlockManager: Removing RDD 0
21/06/04 00:42:28 INFO ReceivedBlockTracker: Deleting batches: 
21/06/04 00:42:28 INFO InputInfoTracker: remove old batch metadata: 
21/06/04 00:42:28 INFO SparkContext: Starting job: count at KafkaSparkIntegrationExample.java:49
21/06/04 00:42:28 INFO DAGScheduler: Got job 6 (count at KafkaSparkIntegrationExample.java:49) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 6 (count at KafkaSparkIntegrationExample.java:49)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 6 (MapPartitionsRDD[5] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_6 stored as values in memory (estimated size 4.4 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_6_piece0 stored as bytes in memory (estimated size 2.7 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_6_piece0 in memory on 192.168.1.7:52779 (size: 2.7 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 6 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 6 (MapPartitionsRDD[5] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 6.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 6.0 (TID 6) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 6.0 (TID 6)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 6.0 (TID 6). 875 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 6.0 (TID 6) in 8 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 6.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 6 (count at KafkaSparkIntegrationExample.java:49) finished in 0.025 s
21/06/04 00:42:28 INFO DAGScheduler: Job 6 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 6: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 6 finished: count at KafkaSparkIntegrationExample.java:49, took 0.032653 s
Count of messages:0
21/06/04 00:42:28 INFO SparkContext: Starting job: foreach at KafkaSparkIntegrationExample.java:50
21/06/04 00:42:28 INFO DAGScheduler: Got job 7 (foreach at KafkaSparkIntegrationExample.java:50) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 7 (foreach at KafkaSparkIntegrationExample.java:50)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 7 (MapPartitionsRDD[5] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_7 stored as values in memory (estimated size 4.9 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_7_piece0 stored as bytes in memory (estimated size 2.8 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_7_piece0 in memory on 192.168.1.7:52779 (size: 2.8 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 7 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 7 (MapPartitionsRDD[5] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 7.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 7.0 (TID 7) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 7.0 (TID 7)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 7.0 (TID 7). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 7.0 (TID 7) in 11 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 7.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 7 (foreach at KafkaSparkIntegrationExample.java:50) finished in 0.025 s
21/06/04 00:42:28 INFO DAGScheduler: Job 7 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 7: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 7 finished: foreach at KafkaSparkIntegrationExample.java:50, took 0.030796 s
21/06/04 00:42:28 INFO JobScheduler: Finished job streaming job 1622747547000 ms.0 from job set of time 1622747547000 ms
21/06/04 00:42:28 INFO JobScheduler: Starting job streaming job 1622747547000 ms.1 from job set of time 1622747547000 ms
21/06/04 00:42:28 INFO SparkContext: Starting job: foreachPartition at KafkaSparkIntegrationExample.java:56
21/06/04 00:42:28 INFO DAGScheduler: Got job 8 (foreachPartition at KafkaSparkIntegrationExample.java:56) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 8 (foreachPartition at KafkaSparkIntegrationExample.java:56)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 8 (KafkaRDD[4] at createDirectStream at KafkaSparkIntegrationExample.java:41), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_8 stored as values in memory (estimated size 3.5 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_8_piece0 stored as bytes in memory (estimated size 2.1 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_8_piece0 in memory on 192.168.1.7:52779 (size: 2.1 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 8 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 8 (KafkaRDD[4] at createDirectStream at KafkaSparkIntegrationExample.java:41) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 8.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 8.0 (TID 8) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 8.0 (TID 8)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 8.0 (TID 8). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 8.0 (TID 8) in 9 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 8.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 8 (foreachPartition at KafkaSparkIntegrationExample.java:56) finished in 0.025 s
21/06/04 00:42:28 INFO DAGScheduler: Job 8 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 8: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 8 finished: foreachPartition at KafkaSparkIntegrationExample.java:56, took 0.030022 s
21/06/04 00:42:28 INFO JobScheduler: Finished job streaming job 1622747547000 ms.1 from job set of time 1622747547000 ms
21/06/04 00:42:28 INFO JobScheduler: Total delay: 1.768 s for time 1622747547000 ms (execution: 0.151 s)
21/06/04 00:42:28 INFO MapPartitionsRDD: Removing RDD 3 from persistence list
21/06/04 00:42:28 INFO JobScheduler: Starting job streaming job 1622747548000 ms.0 from job set of time 1622747548000 ms
21/06/04 00:42:28 INFO KafkaRDD: Removing RDD 2 from persistence list
21/06/04 00:42:28 INFO BlockManager: Removing RDD 3
21/06/04 00:42:28 INFO BlockManager: Removing RDD 2
21/06/04 00:42:28 INFO ReceivedBlockTracker: Deleting batches: 
21/06/04 00:42:28 INFO InputInfoTracker: remove old batch metadata: 1622747545000 ms
21/06/04 00:42:28 INFO SparkContext: Starting job: count at KafkaSparkIntegrationExample.java:49
21/06/04 00:42:28 INFO DAGScheduler: Got job 9 (count at KafkaSparkIntegrationExample.java:49) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 9 (count at KafkaSparkIntegrationExample.java:49)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 9 (MapPartitionsRDD[7] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_9 stored as values in memory (estimated size 4.4 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_9_piece0 stored as bytes in memory (estimated size 2.7 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_9_piece0 in memory on 192.168.1.7:52779 (size: 2.7 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 9 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 9 (MapPartitionsRDD[7] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 9.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 9.0 (TID 9) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 9.0 (TID 9)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 9.0 (TID 9). 875 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 9.0 (TID 9) in 8 ms on 192.168.1.7 (executor driver) (1/1)
21/06/04 00:42:28 INFO TaskSchedulerImpl: Removed TaskSet 9.0, whose tasks have all completed, from pool 
21/06/04 00:42:28 INFO DAGScheduler: ResultStage 9 (count at KafkaSparkIntegrationExample.java:49) finished in 0.023 s
21/06/04 00:42:28 INFO DAGScheduler: Job 9 is finished. Cancelling potential speculative or zombie tasks for this job
21/06/04 00:42:28 INFO TaskSchedulerImpl: Killing all running tasks in stage 9: Stage finished
21/06/04 00:42:28 INFO DAGScheduler: Job 9 finished: count at KafkaSparkIntegrationExample.java:49, took 0.029593 s
Count of messages:0
21/06/04 00:42:28 INFO SparkContext: Starting job: foreach at KafkaSparkIntegrationExample.java:50
21/06/04 00:42:28 INFO DAGScheduler: Got job 10 (foreach at KafkaSparkIntegrationExample.java:50) with 1 output partitions
21/06/04 00:42:28 INFO DAGScheduler: Final stage: ResultStage 10 (foreach at KafkaSparkIntegrationExample.java:50)
21/06/04 00:42:28 INFO DAGScheduler: Parents of final stage: List()
21/06/04 00:42:28 INFO DAGScheduler: Missing parents: List()
21/06/04 00:42:28 INFO DAGScheduler: Submitting ResultStage 10 (MapPartitionsRDD[7] at map at KafkaSparkIntegrationExample.java:46), which has no missing parents
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_10 stored as values in memory (estimated size 4.9 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO MemoryStore: Block broadcast_10_piece0 stored as bytes in memory (estimated size 2.8 KiB, free 366.3 MiB)
21/06/04 00:42:28 INFO BlockManagerInfo: Added broadcast_10_piece0 in memory on 192.168.1.7:52779 (size: 2.8 KiB, free: 366.3 MiB)
21/06/04 00:42:28 INFO SparkContext: Created broadcast 10 from broadcast at DAGScheduler.scala:1388
21/06/04 00:42:28 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 10 (MapPartitionsRDD[7] at map at KafkaSparkIntegrationExample.java:46) (first 15 tasks are for partitions Vector(0))
21/06/04 00:42:28 INFO TaskSchedulerImpl: Adding task set 10.0 with 1 tasks resource profile 0
21/06/04 00:42:28 INFO TaskSetManager: Starting task 0.0 in stage 10.0 (TID 10) (192.168.1.7, executor driver, partition 0, PROCESS_LOCAL, 4357 bytes) taskResourceAssignments Map()
21/06/04 00:42:28 INFO Executor: Running task 0.0 in stage 10.0 (TID 10)
21/06/04 00:42:28 INFO KafkaRDD: Beginning offset 0 is the same as ending offset skipping new_topic 0
21/06/04 00:42:28 INFO Executor: Finished task 0.0 in stage 10.0 (TID 10). 837 bytes result sent to driver
21/06/04 00:42:28 INFO TaskSetManager: Finished task 0.0 in stage 10.0 (TID 10) in 7 ms on 192.168.1.7 (executor driver) (1/1)

The snippet below shows the messages received in the example code.

Snippet – Output

21/06/04 00:42:59 INFO TaskSchedulerImpl: Killing all running tasks in stage 102: Stage finished
21/06/04 00:42:59 INFO DAGScheduler: Job 102 finished: count at KafkaSparkIntegrationExample.java:49, took 0.216731 s
Count of messages:1
21/06/04 00:42:59 INFO InternalKafkaConsumer: Initial fetch for spark-executor-use_a_separate_group_id_for_each_stream_ new_topic-0 0
21/06/04 00:42:59 INFO KafkaConsumer: [Consumer clientId=consumer-spark-executor-use_a_separate_group_id_for_each_stream_-2, groupId=spark-executor-use_a_separate_group_id_for_each_stream_] Seeking to offset 0 for partition new_topic-0
Kafka topic spark factor

21/06/04 00:43:15 INFO In21/06/04 00:43:15 INFO DAGScheduler: Job 150 finished: count at KafkaSparkIntegrationExample.java:49, took 0.028252 s
Count of messages:1
21/06/04 00:43:15 INFO  InternalKafkaConsumer: Initial fetch for spark-executor-use_a_separate_group_id_for_each_stream_ new_topic-0 1
21/06/04 00:43:15 INFO KafkaConsumer: [Consumer clientId=consumer-spark-executor-use_a_separate_group_id_for_each_stream_-2, groupId=spark-executor-use_a_separate_group_id_for_each_stream_] Seeking to offset 1 for partition new_topic-0
zookeeper server message
21/06/04 00:43:17 INFO DAGScheduler: Job 156 finished: count at KafkaSparkIntegrationExample.java:49, took 0.016666 s
Count of messages:1
21/06/04 00:43:17 INFO InternalKafkaConsumer: Initial fetch for spark-executor-use_a_separate_group_id_for_each_stream_ new_topic-0 2
21/06/04 00:43:17 INFO KafkaConsumer: [Consumer clientId=consumer-spark-executor-use_a_separate_group_id_for_each_stream_-2, groupId=spark-executor-use_a_separate_group_id_for_each_stream_] Seeking to offset 2 for partition new_topic-0
this is a test
21/06/04 00:43:31 INFO InternalKafkaConsumer: Initial fetch for spark-executor-use_a_separate_group_id_for_each_stream_ new_topic-0 3
21/06/04 00:43:31 INFO KafkaConsumer: [Consumer clientId=consumer-spark-executor-use_a_separate_group_id_for_each_stream_-2, groupId=spark-executor-use_a_separate_group_id_for_each_stream_] Seeking to offset 3 for partition new_topic-0
scalable architecture

3. Download the Source Code

Download
You can download the full source code of this example here: Apache Kafka Integration With Spark

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