はじめに
Apache Kafkaの商用版であるConfluent Platformについてのメモ書きです。初物なので調べた内容の整理なども含めてログを残しておきます。
商用版といってもConfluent Community Licenseというのがあり、一部の機能は無償で利用できるので無償の範囲で試しています。
関連記事
Confluent Platform メモ - (1)環境構築
Confluent Platform メモ - (2)メッセージ送受信簡易テスト
Confluent Platform メモ - (3)Schema Registry簡易テスト
参考情報
Apache Kafkaの概要とアーキテクチャ
Confluent Platformってどんなの？
Confluent Platform(Community版)をインストールしてKafka Connectを試してみる
Apache KafkaのProducer/Broker/Consumerのしくみと設定一覧
Manual Install using ZIP and TAR Archives
Confluent Platform Community Licenseについて
Confluent Platformにはいくつかコンポーネントが含まれており、コンポーネントごとにライセンスが異なるようです。
Confluent Community License FAQ
以下の表は上のサイトからの抜粋です。
Confluent Platformのコアのコンポーネントである Kafka, ZooKeeper辺りはAPACHE 2.0 LICENCEとして利用可能のようです。Schema RgistryやREST Proxyは COMMUNITY LICENSEの範囲で利用できそうです。
Control Centerやクラスター関連の機能を使う場合にはENTERPRISE LICENSEが必要(有償)ということのようです。
参考:
Confluent Platform Licenses
Confluent Community License Version 1.0
環境情報
RHEL V8.2
Confluent Community V6.2.0
Windows10上のVirtualBoxに仮想OSとしてRHEL V8.2を立てているのでそこにインストールします。
Kafkaは複数のノードに分散されたクラスター構成を行うことができますが、ここでは簡易的なテスト環境を作る目的なので、1ノード上に1Brokerのみを構成します。
前提条件はこちら
Confluent System Requirements
javaは既にインストール済み
[root@test12 ~]# java -version
openjdk version "1.8.0_242"
OpenJDK Runtime Environment (build 1.8.0_242-b08)
OpenJDK 64-Bit Server VM (build 25.242-b08, mixed mode)
Confluent Platform Community Componentのインストール
オフラインでのインストール方法があるのでこちらの手順に従ってやってみます。
Manual Install using ZIP and TAR Archives
上のガイドにある confluent-community-6.2.0.tar.gz (約350MB) をダウンロードしてターゲットマシンに持っていき展開します。ここでは、/opt/以下に展開することにします。
[root@test12 /Local_Inst_Image/Confluent]# ls -la
合計 348552
drwxrwx---. 1 root vboxsf 0 8月 13 18:00 .
drwxrwx---. 1 root vboxsf 16384 8月 13 17:59 ..
-rwxrwx---. 1 root vboxsf 356898902 8月 13 18:00 confluent-community-6.2.0.tar.gz
[root@test12 /Local_Inst_Image/Confluent]# tar xzf confluent-community-6.2.0.tar.gz -C /opt
/opt/confluent-6.2.0以下にファイルが展開されました。
owner,group変更します。
[root@test12 /opt]# chown -R root:root confluent-6.2.0
[root@test12 /opt]# ls -la confluent-6.2.0/
合計 8
drwxr-xr-x. 7 root root 77 6月 6 08:51 .
drwxr-xr-x. 7 root root 95 8月 13 18:05 ..
-rw-r--r--. 1 root root 871 6月 6 08:51 README
drwxr-xr-x. 3 root root 4096 6月 6 07:11 bin
drwxr-xr-x. 8 root root 116 6月 6 07:11 etc
drwxr-xr-x. 3 root root 21 6月 6 07:11 lib
drwxr-xr-x. 6 root root 71 6月 6 07:11 share
drwxr-xr-x. 2 root root 178 6月 6 08:51 src
tar展開するだけでOKみたいです。楽ちん。
Confluentコンポーネントの管理
ここでは、ZooKeeper, Kafka(Broker), Schem Registry の3つのコンポーネントを扱います。
とりあえず全てデフォルトの構成で起動させてみます。
(Control Centerというコンポーネントを使うとGUIの管理用インターフェースが提供されるようで、それを使うと分かりやすそうなのですが、残念ながらCommunity Licenseでは利用できないようです...)
ZooKeeper
構成
Configure Confluent Platform - ZooKeeper
デフォルトで提供されているプロパティファイルを確認。
/opt/confluent-6.2.0/etc/kafka/zookeeper.properties
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# the directory where the snapshot is stored.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080
この辺にパラメーターのReferenceらしきものはあり...
Running ZooKeeper in Production - Configuration Options
Complete Listはこっち見ろということでApacheのサイトに飛ばされたが...
Apache - ZooKeeper Administrator's Guide
admin.enableServerとか見つからない。上のリンクはV3.4.10のドキュメントのようだがバージョンが古いっぽい。
新し目のバージョンのドキュメントにはadmin.enableServerの記載はあった。
Apache - ZooKeeper Administrator's Guide (3.5.9)
(後でZooKeeperの起動時ログを確認したところV3.5.9だった)
起動/停止
- 起動コマンド:
bin/zookeeper-server-start -daemon <property_file> - 停止コマンド:
bin/zookeeper-server-stop - ログ:
logs/zookeeper.out - Listenポート: 2181
※マニュアルにコマンドリファレンスみたいなものが見当たらない...。上のファイル覗いたらシェル・スクリプトだったのでそこから判断するに -daemonオプションをつければバックグラウンド起動してくれるようです。(末尾のシェル・スクリプト一覧参照)
提供されるプロパティファイルをそのまま使って以下のコマンドで起動してみます。
/opt/confluent-6.2.0/bin/zookeeper-server-start -daemon /opt/confluent-6.2.0/etc/kafka/zookeeper.properties
logs/zookeeper.outにメッセージが出力されます。
(server.logにも同じ内容のメッセージが出力されますが、server.logは後述のKafkaのメッセージも合わせて出力されるようです。)
起動時ログ: logs/zookeeper.out
[2021-08-14 09:24:44,670] INFO Reading configuration from: /opt/confluent-6.2.0/etc/kafka/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-08-14 09:24:44,695] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-08-14 09:24:44,695] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-08-14 09:24:44,697] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-08-14 09:24:44,697] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-08-14 09:24:44,697] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2021-08-14 09:24:44,697] WARN Either no config or no quorum defined in config, running in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2021-08-14 09:24:44,700] INFO Log4j 1.2 jmx support found and enabled. (org.apache.zookeeper.jmx.ManagedUtil)
[2021-08-14 09:24:44,714] INFO Reading configuration from: /opt/confluent-6.2.0/etc/kafka/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-08-14 09:24:44,714] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-08-14 09:24:44,714] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2021-08-14 09:24:44,715] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2021-08-14 09:24:44,717] INFO zookeeper.snapshot.trust.empty : false (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2021-08-14 09:24:44,734] INFO Server environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,734] INFO Server environment:host.name=localhost (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,734] INFO Server environment:java.version=1.8.0_242 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,734] INFO Server environment:java.vendor=Oracle Corporation (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,734] INFO Server environment:java.home=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-4.el8.x86_64/jre (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,735] INFO Server environment:java.class.path=/opt/confluent-6.2.0/bin/../share/java/kafka/netty-transport-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-buffer-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-mirror-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-scala_2.13-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-util-ajax-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.inject-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/audience-annotations-0.5.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-java8-compat_2.13-0.9.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-metadata-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.activation-api-1.2.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/slf4j-api-1.7.30.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-logging_2.13-3.9.2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/zookeeper-jute-3.5.9.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-codec-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-json-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-security-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-reflect-2.13.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-library-2.13.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-jaxrs-base-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-sources.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-handler-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.ws.rs-api-2.1.6.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/zstd-jni-1.4.9-1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-util-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-server-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/zookeeper-3.5.9.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/javassist-3.27.0-GA.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-shell-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-client-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/commons-cli-1.4.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-core-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-servlet-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-runtime-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-collection-compat_2.13-2.3.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-jaxrs-json-provider-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.validation-api-2.0.2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-module-scala_2.13-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.annotation-api-1.3.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-server-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/hk2-utils-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/lz4-java-1.7.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/maven-artifact-3.8.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-transforms-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/metrics-core-2.2.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-annotations-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-test-sources.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-mirror-client-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.xml.bind-api-2.3.2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/reflections-0.9.12.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/argparse4j-0.7.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-transport-native-epoll-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/activation-1.1.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-raft-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-tools-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-http-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-javadoc.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-api-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-file-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-container-servlet-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-container-servlet-core-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-resolver-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jaxb-api-2.3.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-test-utils-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-log4j-appender-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-hk2-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/javax.servlet-api-3.1.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-datatype-jdk8-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/slf4j-log4j12-1.7.30.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/commons-lang3-3.8.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/plexus-utils-3.2.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/javax.ws.rs-api-2.1.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-common-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-module-paranamer-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-common-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-transport-native-unix-common-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-test.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-examples-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-io-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/osgi-resource-locator-1.0.3.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/rocksdbjni-5.18.4.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/aopalliance-repackaged-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-dataformat-csv-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/snappy-java-1.1.8.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-databind-2.10.5.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-continuation-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/hk2-locator-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/confluent-log4j-1.2.17-cp2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/hk2-api-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jopt-simple-5.0.4.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/paranamer-2.8.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-module-jaxb-annotations-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-clients-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-servlets-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-basic-auth-extension-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jline-3.12.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-client-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/confluent-telemetry/* (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,737] INFO Server environment:java.library.path=/opt/ibm/cics/lib:/opt/ibm/cicssm/lib:/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,737] INFO Server environment:java.io.tmpdir=/tmp (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,737] INFO Server environment:java.compiler=<NA> (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,737] INFO Server environment:os.name=Linux (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:os.arch=amd64 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:os.version=4.18.0-193.el8.x86_64 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:user.name=root (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:user.home=/root (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:user.dir=/opt/confluent-6.2.0/bin (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:os.memory.free=497MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:os.memory.max=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,738] INFO Server environment:os.memory.total=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,740] INFO minSessionTimeout set to 6000 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,740] INFO maxSessionTimeout set to 60000 (org.apache.zookeeper.server.ZooKeeperServer)
[2021-08-14 09:24:44,740] 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-08-14 09:24:44,749] INFO Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory (org.apache.zookeeper.server.ServerCnxnFactory)
[2021-08-14 09:24:44,753] 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-08-14 09:24:44,757] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
[2021-08-14 09:24:44,800] INFO zookeeper.snapshotSizeFactor = 0.33 (org.apache.zookeeper.server.ZKDatabase)
[2021-08-14 09:24:44,803] INFO Reading snapshot /tmp/zookeeper/version-2/snapshot.0 (org.apache.zookeeper.server.persistence.FileSnap)
[2021-08-14 09:24:44,810] INFO Snapshotting: 0x0 to /tmp/zookeeper/version-2/snapshot.0 (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2021-08-14 09:24:44,822] INFO PrepRequestProcessor (sid:0) started, reconfigEnabled=false (org.apache.zookeeper.server.PrepRequestProcessor)
[2021-08-14 09:24:44,828] INFO Using checkIntervalMs=60000 maxPerMinute=10000 (org.apache.zookeeper.server.ContainerManager)
起動したっぽい。ポート2181がListenされた状態になりました。
起動時のログ見ると以下のメッセージが出てたので、ZooKeeperのバージョンとしては3.5.9らしい。
Server environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.server.ZooKeeperServer)
Kafka(Broker)
構成
Configure Confluent Platform - Kafka
製品提供のプロパティファイルはこちら。
/opt/confluent-6.2.0/etc/kafka/server.properties
/opt/confluent-6.2.0/etc/kafka/server.properties
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# see kafka.server.KafkaConfig for additional details and defaults
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092
# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092
# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3
# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1
############################# Internal Topic Settings #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Flush Policy #############################
# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
# 1. Durability: Unflushed data may be lost if you are not using replication.
# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.
# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=18000
##################### Confluent Metrics Reporter #######################
# Confluent Control Center and Confluent Auto Data Balancer integration
#
# Uncomment the following lines to publish monitoring data for
# Confluent Control Center and Confluent Auto Data Balancer
# If you are using a dedicated metrics cluster, also adjust the settings
# to point to your metrics kakfa cluster.
#metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
#confluent.metrics.reporter.bootstrap.servers=localhost:9092
#
# Uncomment the following line if the metrics cluster has a single broker
#confluent.metrics.reporter.topic.replicas=1
############################# Group Coordinator Settings #############################
# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0
コメントが多くて長いのでパラメータ部分だけ抽出するとこんな感じ。
[root@test12 /opt/confluent-6.2.0/etc/kafka]# cat server.properties | grep -v -e "^#" | grep -v -e "^\s*$"
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
パラメーターのリファレンスはこちら
Kafka Broker Configurations
※Listenするポート関連の設定がされていないが、リファレンスを見るとlisnersが指定されていなければportが使われるっぽく、portのデフォルトが9092になってました。
※topicに保持されるデータは実体としてはOS上のファイルとして保持されます。その書き出し先はlog.dirsパラメーターで指定され、上の通りデフォルトでは/tmp/kafka-logsになっています。/tmp/ディレクトリ以下は通常は自動クリーンアップの対象になっていることが多いので、自動削除されないようにするにはこのパラメーターの設定を変更して出力先ディレクトリを変更するか、/tmpクリーンアップの設定を変更する必要があります。
参考:
Stack Overflow - Which directory does apache kafka store the data in broker nodes
Kafka Broker Configurations - log.dirs
【CentOS】/tmp配下のファイルが消える理由
起動/停止
- 起動コマンド:
bin/kafka-server-start -daemon <property_file> - 停止コマンド:
bin/kafka-server-stop - ログ:
logs/kafkaServer.out - Listenポート: 9092
以下のコマンドで起動
/opt/confluent-6.2.0/bin/kafka-server-start -daemon /opt/confluent-6.2.0/etc/kafka/server.properties
起動時ログ: logs/kafkaServer.out
[2021-08-14 11:17:16,090] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2021-08-14 11:17:16,584] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2021-08-14 11:17:16,712] INFO Registered signal handlers for TERM, INT, HUP (org.apache.kafka.common.utils.LoggingSignalHandler)
[2021-08-14 11:17:16,716] INFO starting (kafka.server.KafkaServer)
[2021-08-14 11:17:16,717] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
[2021-08-14 11:17:16,747] INFO [ZooKeeperClient Kafka server] Initializing a new session to localhost:2181. (kafka.zookeeper.ZooKeeperClient)
[2021-08-14 11:17:16,752] INFO Client environment:zookeeper.version=3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 20:03 GMT (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,752] INFO Client environment:host.name=localhost (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,752] INFO Client environment:java.version=1.8.0_242 (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,752] INFO Client environment:java.vendor=Oracle Corporation (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,752] INFO Client environment:java.home=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-4.el8.x86_64/jre (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,761] INFO Client environment:java.class.path=/opt/confluent-6.2.0/bin/../share/java/kafka/netty-transport-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-buffer-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-mirror-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-scala_2.13-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-util-ajax-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.inject-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/audience-annotations-0.5.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-java8-compat_2.13-0.9.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-metadata-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.activation-api-1.2.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/slf4j-api-1.7.30.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-logging_2.13-3.9.2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/zookeeper-jute-3.5.9.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-codec-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-json-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-security-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-reflect-2.13.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-library-2.13.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-jaxrs-base-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-sources.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-handler-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.ws.rs-api-2.1.6.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/zstd-jni-1.4.9-1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-util-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-server-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/zookeeper-3.5.9.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/javassist-3.27.0-GA.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-shell-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-client-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/commons-cli-1.4.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-core-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-servlet-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-runtime-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/scala-collection-compat_2.13-2.3.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-jaxrs-json-provider-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.validation-api-2.0.2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-module-scala_2.13-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.annotation-api-1.3.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-server-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/hk2-utils-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/lz4-java-1.7.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/maven-artifact-3.8.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-transforms-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/metrics-core-2.2.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-annotations-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-test-sources.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-mirror-client-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jakarta.xml.bind-api-2.3.2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/reflections-0.9.12.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/argparse4j-0.7.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-transport-native-epoll-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/activation-1.1.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-raft-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-tools-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-http-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-javadoc.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-api-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-file-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-container-servlet-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-container-servlet-core-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-resolver-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jaxb-api-2.3.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-test-utils-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-log4j-appender-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-hk2-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/javax.servlet-api-3.1.0.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-datatype-jdk8-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/slf4j-log4j12-1.7.30.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/commons-lang3-3.8.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/plexus-utils-3.2.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/javax.ws.rs-api-2.1.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-common-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-module-paranamer-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-common-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/netty-transport-native-unix-common-4.1.62.Final.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka_2.13-6.2.0-ccs-test.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-streams-examples-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-io-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/osgi-resource-locator-1.0.3.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/rocksdbjni-5.18.4.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/aopalliance-repackaged-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-dataformat-csv-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/snappy-java-1.1.8.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-databind-2.10.5.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-continuation-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/hk2-locator-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/confluent-log4j-1.2.17-cp2.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/hk2-api-2.6.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jopt-simple-5.0.4.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/paranamer-2.8.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jackson-module-jaxb-annotations-2.10.5.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/kafka-clients-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jetty-servlets-9.4.40.v20210413.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/connect-basic-auth-extension-6.2.0-ccs.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jline-3.12.1.jar:/opt/confluent-6.2.0/bin/../share/java/kafka/jersey-client-2.34.jar:/opt/confluent-6.2.0/bin/../share/java/confluent-telemetry/* (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,763] INFO Client environment:java.library.path=/opt/ibm/cics/lib:/opt/ibm/cicssm/lib:/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,763] INFO Client environment:java.io.tmpdir=/tmp (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,763] INFO Client environment:java.compiler=<NA> (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,763] INFO Client environment:os.name=Linux (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,763] INFO Client environment:os.arch=amd64 (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,763] INFO Client environment:os.version=4.18.0-193.el8.x86_64 (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,764] INFO Client environment:user.name=root (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,764] INFO Client environment:user.home=/root (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,764] INFO Client environment:user.dir=/opt/confluent-6.2.0/bin (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,764] INFO Client environment:os.memory.free=980MB (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,764] INFO Client environment:os.memory.max=1024MB (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,764] INFO Client environment:os.memory.total=1024MB (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,766] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=18000 watcher=kafka.zookeeper.ZooKeeperClient$ZooKeeperClientWatcher$@5b8dfcc1 (org.apache.zookeeper.ZooKeeper)
[2021-08-14 11:17:16,770] INFO jute.maxbuffer value is 4194304 Bytes (org.apache.zookeeper.ClientCnxnSocket)
[2021-08-14 11:17:16,785] INFO zookeeper.request.timeout value is 0. feature enabled= (org.apache.zookeeper.ClientCnxn)
[2021-08-14 11:17:16,796] INFO [ZooKeeperClient Kafka server] Waiting until connected. (kafka.zookeeper.ZooKeeperClient)
[2021-08-14 11:17:16,798] INFO Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2021-08-14 11:17:16,802] INFO Socket connection established, initiating session, client: /0:0:0:0:0:0:0:1:37008, server: localhost/0:0:0:0:0:0:0:1:2181 (org.apache.zookeeper.ClientCnxn)
[2021-08-14 11:17:16,831] INFO Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x1000070216a0000, negotiated timeout = 18000 (org.apache.zookeeper.ClientCnxn)
[2021-08-14 11:17:16,834] INFO [ZooKeeperClient Kafka server] Connected. (kafka.zookeeper.ZooKeeperClient)
[2021-08-14 11:17:17,053] INFO [feature-zk-node-event-process-thread]: Starting (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)
[2021-08-14 11:17:17,065] INFO Feature ZK node at path: /feature does not exist (kafka.server.FinalizedFeatureChangeListener)
[2021-08-14 11:17:17,065] INFO Cleared cache (kafka.server.FinalizedFeatureCache)
[2021-08-14 11:17:17,311] INFO Cluster ID = Mp9ekWqlR2-jIyzo3tDKMg (kafka.server.KafkaServer)
[2021-08-14 11:17:17,314] WARN No meta.properties file under dir /tmp/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2021-08-14 11:17:17,381] 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-08-14 11:17:17,394] 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-08-14 11:17:17,454] INFO [ThrottledChannelReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-08-14 11:17:17,455] INFO [ThrottledChannelReaper-Produce]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-08-14 11:17:17,456] INFO [ThrottledChannelReaper-Request]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-08-14 11:17:17,458] INFO [ThrottledChannelReaper-ControllerMutation]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
[2021-08-14 11:17:17,484] INFO Log directory /tmp/kafka-logs not found, creating it. (kafka.log.LogManager)
[2021-08-14 11:17:17,500] INFO Loading logs from log dirs ArraySeq(/tmp/kafka-logs) (kafka.log.LogManager)
[2021-08-14 11:17:17,506] INFO Attempting recovery for all logs in /tmp/kafka-logs since no clean shutdown file was found (kafka.log.LogManager)
[2021-08-14 11:17:17,532] INFO Loaded 0 logs in 33ms. (kafka.log.LogManager)
[2021-08-14 11:17:17,533] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
[2021-08-14 11:17:17,536] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
[2021-08-14 11:17:18,486] INFO Updated connection-accept-rate max connection creation rate to 2147483647 (kafka.network.ConnectionQuotas)
[2021-08-14 11:17:18,490] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.Acceptor)
[2021-08-14 11:17:18,568] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Created data-plane acceptor and processors for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-08-14 11:17:18,614] INFO [broker-0-to-controller-send-thread]: Starting (kafka.server.BrokerToControllerRequestThread)
[2021-08-14 11:17:18,647] INFO [ExpirationReaper-0-Produce]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,648] INFO [ExpirationReaper-0-Fetch]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,648] INFO [ExpirationReaper-0-DeleteRecords]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,649] INFO [ExpirationReaper-0-ElectLeader]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,670] INFO [LogDirFailureHandler]: Starting (kafka.server.ReplicaManager$LogDirFailureHandler)
[2021-08-14 11:17:18,701] INFO Creating /brokers/ids/0 (is it secure? false) (kafka.zk.KafkaZkClient)
[2021-08-14 11:17:18,720] INFO Stat of the created znode at /brokers/ids/0 is: 25,25,1628907438713,1628907438713,1,0,0,72058075634860032,202,0,25
(kafka.zk.KafkaZkClient)
[2021-08-14 11:17:18,721] INFO Registered broker 0 at path /brokers/ids/0 with addresses: PLAINTEXT://localhost:9092, czxid (broker epoch): 25 (kafka.zk.KafkaZkClient)
[2021-08-14 11:17:18,788] INFO [ExpirationReaper-0-topic]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,799] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,801] INFO [ExpirationReaper-0-Rebalance]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,803] INFO Successfully created /controller_epoch with initial epoch 0 (kafka.zk.KafkaZkClient)
[2021-08-14 11:17:18,815] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.group.GroupCoordinator)
[2021-08-14 11:17:18,849] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.group.GroupCoordinator)
[2021-08-14 11:17:18,850] INFO Feature ZK node created at path: /feature (kafka.server.FinalizedFeatureChangeListener)
[2021-08-14 11:17:18,897] INFO Updated cache from existing <empty> to latest FinalizedFeaturesAndEpoch(features=Features{}, epoch=0). (kafka.server.FinalizedFeatureCache)
[2021-08-14 11:17:18,902] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1 (kafka.coordinator.transaction.ProducerIdManager)
[2021-08-14 11:17:18,902] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2021-08-14 11:17:18,906] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2021-08-14 11:17:18,908] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2021-08-14 11:17:18,946] INFO [ExpirationReaper-0-AlterAcls]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-08-14 11:17:18,974] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2021-08-14 11:17:19,052] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Starting socket server acceptors and processors (kafka.network.SocketServer)
[2021-08-14 11:17:19,069] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-08-14 11:17:19,070] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started socket server acceptors and processors (kafka.network.SocketServer)
[2021-08-14 11:17:19,074] INFO Kafka version: 6.2.0-ccs (org.apache.kafka.common.utils.AppInfoParser)
[2021-08-14 11:17:19,074] INFO Kafka commitId: 1a5755cf9401c84f (org.apache.kafka.common.utils.AppInfoParser)
[2021-08-14 11:17:19,074] INFO Kafka startTimeMs: 1628907439070 (org.apache.kafka.common.utils.AppInfoParser)
[2021-08-14 11:17:19,076] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
[2021-08-14 11:17:19,134] 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)
ポート9092がListenされた状態になりました。
Schema Registry
構成
Configure Confluent Platform - Schema Registry
製品提供のプロパティファイルはこちら。
etc/schema-registry/schema-registry.properties
/opt/confluent-6.2.0/etc/schema-registry/schema-registry.properties
#
# Copyright 2018 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# The address the socket server listens on.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=http://0.0.0.0:8081
# Zookeeper connection string for the Zookeeper cluster used by your Kafka cluster
# (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# Note: use of this property is deprecated.
#kafkastore.connection.url=localhost:2181
# Alternatively, Schema Registry can now operate without Zookeeper, handling all coordination via
# Kafka brokers. Use this setting to specify the bootstrap servers for your Kafka cluster and it
# will be used both for selecting the leader schema registry instance and for storing the data for
# registered schemas.
# (Note that you cannot mix the two modes; use this mode only on new deployments or by shutting down
# all instances, switching to the new configuration, and then starting the schema registry
# instances again.)
kafkastore.bootstrap.servers=PLAINTEXT://localhost:9092
# The name of the topic to store schemas in
kafkastore.topic=_schemas
# If true, API requests that fail will include extra debugging information, including stack traces
debug=false
パラメーターのリファレンスはこちら
Schema Registry Configuration Options
起動/停止
- 起動コマンド:
bin/schema-registry-start -daemon <property_file> - 停止コマンド:
bin/schema-registry-stop - ログ:
logs/schema-registry.log - Listenポート: 8081
以下のコマンドで起動
/opt/confluent-6.2.0/bin/schema-registry-start -daemon /opt/confluent-6.2.0/etc/schema-registry/schema-registry.properties
起動時ログ: logs/schema-registry.log
[2021-08-14 11:57:16,923] INFO SchemaRegistryConfig values:
access.control.allow.headers =
access.control.allow.methods =
access.control.allow.origin =
access.control.skip.options = true
authentication.method = NONE
authentication.realm =
authentication.roles = [*]
authentication.skip.paths = []
avro.compatibility.level =
compression.enable = true
csrf.prevention.enable = false
csrf.prevention.token.endpoint = /csrf
csrf.prevention.token.expiration.minutes = 30
csrf.prevention.token.max.entries = 10000
debug = false
host.name = localhost
idle.timeout.ms = 30000
inter.instance.headers.whitelist = []
inter.instance.protocol = http
kafkastore.bootstrap.servers = [PLAINTEXT://localhost:9092]
kafkastore.checkpoint.dir = /tmp
kafkastore.checkpoint.version = 0
kafkastore.connection.url =
kafkastore.group.id =
kafkastore.init.timeout.ms = 60000
kafkastore.sasl.kerberos.kinit.cmd = /usr/bin/kinit
kafkastore.sasl.kerberos.min.time.before.relogin = 60000
kafkastore.sasl.kerberos.service.name =
kafkastore.sasl.kerberos.ticket.renew.jitter = 0.05
kafkastore.sasl.kerberos.ticket.renew.window.factor = 0.8
kafkastore.sasl.mechanism = GSSAPI
kafkastore.security.protocol = PLAINTEXT
kafkastore.ssl.cipher.suites =
kafkastore.ssl.enabled.protocols = TLSv1.2,TLSv1.1,TLSv1
kafkastore.ssl.endpoint.identification.algorithm =
kafkastore.ssl.key.password = [hidden]
kafkastore.ssl.keymanager.algorithm = SunX509
kafkastore.ssl.keystore.location =
kafkastore.ssl.keystore.password = [hidden]
kafkastore.ssl.keystore.type = JKS
kafkastore.ssl.protocol = TLS
kafkastore.ssl.provider =
kafkastore.ssl.trustmanager.algorithm = PKIX
kafkastore.ssl.truststore.location =
kafkastore.ssl.truststore.password = [hidden]
kafkastore.ssl.truststore.type = JKS
kafkastore.timeout.ms = 500
kafkastore.topic = _schemas
kafkastore.topic.replication.factor = 3
kafkastore.topic.skip.validation = false
kafkastore.update.handlers = []
kafkastore.write.max.retries = 5
kafkastore.zk.session.timeout.ms = 30000
leader.eligibility = true
listeners = [http://0.0.0.0:8081]
master.eligibility = null
metric.reporters = []
metrics.jmx.prefix = kafka.schema.registry
metrics.num.samples = 2
metrics.sample.window.ms = 30000
metrics.tag.map = []
mode.mutability = true
port = 8081
request.logger.name = io.confluent.rest-utils.requests
request.queue.capacity = 2147483647
request.queue.capacity.growby = 64
request.queue.capacity.init = 128
resource.extension.class = []
resource.extension.classes = []
resource.static.locations = []
response.http.headers.config =
response.mediatype.default = application/vnd.schemaregistry.v1+json
response.mediatype.preferred = [application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json]
rest.servlet.initializor.classes = []
schema.cache.expiry.secs = 300
schema.cache.size = 1000
schema.compatibility.level = backward
schema.providers = []
schema.registry.group.id = schema-registry
schema.registry.inter.instance.protocol =
schema.registry.resource.extension.class = []
schema.registry.zk.namespace = schema_registry
shutdown.graceful.ms = 1000
ssl.cipher.suites = []
ssl.client.auth = false
ssl.client.authentication = NONE
ssl.enabled.protocols = []
ssl.endpoint.identification.algorithm = null
ssl.key.password = [hidden]
ssl.keymanager.algorithm =
ssl.keystore.location =
ssl.keystore.password = [hidden]
ssl.keystore.reload = false
ssl.keystore.type = JKS
ssl.keystore.watch.location =
ssl.protocol = TLS
ssl.provider =
ssl.trustmanager.algorithm =
ssl.truststore.location =
ssl.truststore.password = [hidden]
ssl.truststore.type = JKS
thread.pool.max = 200
thread.pool.min = 8
websocket.path.prefix = /ws
websocket.servlet.initializor.classes = []
zookeeper.set.acl = false
(io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig)
[2021-08-14 11:57:16,998] INFO Logging initialized @1039ms to org.eclipse.jetty.util.log.Slf4jLog (org.eclipse.jetty.util.log)
[2021-08-14 11:57:17,007] INFO Initial capacity 128, increased by 64, maximum capacity 2147483647. (io.confluent.rest.ApplicationServer)
[2021-08-14 11:57:17,117] INFO Adding listener: http://0.0.0.0:8081 (io.confluent.rest.ApplicationServer)
[2021-08-14 11:57:17,999] INFO Registering schema provider for AVRO: io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider (io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry)
[2021-08-14 11:57:17,999] INFO Registering schema provider for JSON: io.confluent.kafka.schemaregistry.json.JsonSchemaProvider (io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry)
[2021-08-14 11:57:17,999] INFO Registering schema provider for PROTOBUF: io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider (io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry)
[2021-08-14 11:57:18,063] INFO Initializing KafkaStore with broker endpoints: PLAINTEXT://localhost:9092 (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:18,088] INFO Creating schemas topic _schemas (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:18,090] WARN Creating the schema topic _schemas using a replication factor of 1, which is less than the desired one of 3. If this is a production environment, it's crucial to add more brokers and increase the replication factor of the topic. (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:18,442] INFO Kafka store reader thread starting consumer (io.confluent.kafka.schemaregistry.storage.KafkaStoreReaderThread)
[2021-08-14 11:57:18,603] INFO Seeking to beginning for all partitions (io.confluent.kafka.schemaregistry.storage.KafkaStoreReaderThread)
[2021-08-14 11:57:18,604] INFO Initialized last consumed offset to -1 (io.confluent.kafka.schemaregistry.storage.KafkaStoreReaderThread)
[2021-08-14 11:57:18,606] INFO [kafka-store-reader-thread-_schemas]: Starting (io.confluent.kafka.schemaregistry.storage.KafkaStoreReaderThread)
[2021-08-14 11:57:18,851] INFO Wait to catch up until the offset at 0 (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:18,963] INFO Reached offset at 0 (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:18,964] INFO Joining schema registry with Kafka-based coordination (io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry)
[2021-08-14 11:57:19,843] INFO Finished rebalance with leader election result: Assignment{version=1, error=0, leader='sr-1-76d7d323-be6d-42fa-b17e-d6d60fd14ac6', leaderIdentity=version=1,host=localhost,port=8081,scheme=http,leaderEligibility=true} (io.confluent.kafka.schemaregistry.leaderelector.kafka.KafkaGroupLeaderElector)
[2021-08-14 11:57:19,877] INFO Wait to catch up until the offset at 1 (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:19,881] INFO Reached offset at 1 (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2021-08-14 11:57:20,084] INFO jetty-9.4.40.v20210413; built: 2021-04-13T20:42:42.668Z; git: b881a572662e1943a14ae12e7e1207989f218b74; jvm 1.8.0_242-b08 (org.eclipse.jetty.server.Server)
[2021-08-14 11:57:20,185] INFO DefaultSessionIdManager workerName=node0 (org.eclipse.jetty.server.session)
[2021-08-14 11:57:20,186] INFO No SessionScavenger set, using defaults (org.eclipse.jetty.server.session)
[2021-08-14 11:57:20,187] INFO node0 Scavenging every 600000ms (org.eclipse.jetty.server.session)
[2021-08-14 11:57:20,992] INFO HV000001: Hibernate Validator 6.1.7.Final (org.hibernate.validator.internal.util.Version)
[2021-08-14 11:57:21,380] INFO Started o.e.j.s.ServletContextHandler@3e7dd664{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler)
[2021-08-14 11:57:21,401] INFO Started o.e.j.s.ServletContextHandler@71c27ee8{/ws,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler)
[2021-08-14 11:57:21,431] INFO Started NetworkTrafficServerConnector@4c762604{HTTP/1.1, (http/1.1)}{0.0.0.0:8081} (org.eclipse.jetty.server.AbstractConnector)
[2021-08-14 11:57:21,432] INFO Started @5478ms (org.eclipse.jetty.server.Server)
[2021-08-14 11:57:21,432] INFO Server started, listening for requests... (io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain)
ポート8081がListenされた状態になりました。
参考: 起動/停止関連スクリプト
/opt/confluent-6.2.0/bin/zookeeper-server-start
zookeeper-server-start
#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
if [ $# -lt 1 ];
then
echo "USAGE: $0 [-daemon] zookeeper.properties"
exit 1
fi
base_dir=$(dirname $0)
if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
LOG4J_CONFIG_NORMAL_INSTALL="/etc/kafka/log4j.properties"
LOG4J_CONFIG_ZIP_INSTALL="$base_dir/../etc/kafka/log4j.properties"
if [ -e "$LOG4J_CONFIG_NORMAL_INSTALL" ]; then # Normal install layout
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_CONFIG_NORMAL_INSTALL}"
elif [ -e "${LOG4J_CONFIG_ZIP_INSTALL}" ]; then # Simple zip file layout
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_CONFIG_ZIP_INSTALL}"
else # Fallback to normal default
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi
fi
export KAFKA_LOG4J_OPTS
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M"
fi
EXTRA_ARGS=${EXTRA_ARGS-'-name zookeeper -loggc'}
COMMAND=$1
case $COMMAND in
-daemon)
EXTRA_ARGS="-daemon "$EXTRA_ARGS
shift
;;
*)
;;
esac
exec $base_dir/kafka-run-class $EXTRA_ARGS org.apache.zookeeper.server.quorum.QuorumPeerMain "$@"
/opt/confluent-6.2.0/bin/zookeeper-server-stop
zookeeper-server-stop
#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
SIGNAL=${SIGNAL:-TERM}
OSNAME=$(uname -s)
if [[ "$OSNAME" == "OS/390" ]]; then
if [ -z $JOBNAME ]; then
JOBNAME="ZKEESTRT"
fi
PIDS=$(ps -A -o pid,jobname,comm | grep -i $JOBNAME | grep java | grep -v grep | awk '{print $1}')
elif [[ "$OSNAME" == "OS400" ]]; then
PIDS=$(ps -Af | grep java | grep -i QuorumPeerMain | grep -v grep | awk '{print $2}')
else
PIDS=$(ps ax | grep java | grep -i QuorumPeerMain | grep -v grep | awk '{print $1}')
fi
if [ -z "$PIDS" ]; then
echo "No zookeeper server to stop"
exit 1
else
kill -s $SIGNAL $PIDS
fi
/opt/confluent-6.2.0/bin/kafka-server-start
kafka-server-start
#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
if [ $# -lt 1 ];
then
echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
exit 1
fi
base_dir=$(dirname $0)
if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
LOG4J_CONFIG_NORMAL_INSTALL="/etc/kafka/log4j.properties"
LOG4J_CONFIG_ZIP_INSTALL="$base_dir/../etc/kafka/log4j.properties"
if [ -e "$LOG4J_CONFIG_NORMAL_INSTALL" ]; then # Normal install layout
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_CONFIG_NORMAL_INSTALL}"
elif [ -e "${LOG4J_CONFIG_ZIP_INSTALL}" ]; then # Simple zip file layout
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_CONFIG_ZIP_INSTALL}"
else # Fallback to normal default
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi
fi
export KAFKA_LOG4J_OPTS
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi
EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}
COMMAND=$1
case $COMMAND in
-daemon)
EXTRA_ARGS="-daemon "$EXTRA_ARGS
shift
;;
*)
;;
esac
exec $base_dir/kafka-run-class $EXTRA_ARGS kafka.Kafka "$@"
/opt/confluent-6.2.0/bin/kafka-server-stop
kafka-server-stop
#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
SIGNAL=${SIGNAL:-TERM}
OSNAME=$(uname -s)
if [[ "$OSNAME" == "OS/390" ]]; then
if [ -z $JOBNAME ]; then
JOBNAME="KAFKSTRT"
fi
PIDS=$(ps -A -o pid,jobname,comm | grep -i $JOBNAME | grep java | grep -v grep | awk '{print $1}')
elif [[ "$OSNAME" == "OS400" ]]; then
PIDS=$(ps -Af | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $2}')
else
PIDS=$(ps ax | grep ' kafka\.Kafka ' | grep java | grep -v grep | awk '{print $1}')
PIDS_SUPPORT=$(ps ax | grep -i 'io\.confluent\.support\.metrics\.SupportedKafka' | grep java | grep -v grep | awk '{print $1}')
fi
if [ -z "$PIDS" ]; then
# Normal Kafka is not running, but maybe we are running the support wrapper?
if [ -z "${PIDS_SUPPORT}" ]; then
echo "No kafka server to stop"
exit 1
else
kill -s $SIGNAL $PIDS_SUPPORT
fi
else
kill -s $SIGNAL $PIDS
fi
/opt/confluent-6.2.0/bin/kafka-run-class
kafka-run-class
#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
if [ $# -lt 1 ];
then
echo "USAGE: $0 [-daemon] [-name servicename] [-loggc] classname [opts]"
exit 1
fi
# CYGWIN == 1 if Cygwin is detected, else 0.
if [[ $(uname -a) =~ "CYGWIN" ]]; then
CYGWIN=1
else
CYGWIN=0
fi
if [ -z "$INCLUDE_TEST_JARS" ]; then
INCLUDE_TEST_JARS=false
fi
# Exclude jars not necessary for running commands.
regex="(-(test|test-sources|src|scaladoc|javadoc)\.jar|jar.asc)$"
should_include_file() {
if [ "$INCLUDE_TEST_JARS" = true ]; then
return 0
fi
file=$1
if [ -z "$(echo "$file" | egrep "$regex")" ] ; then
return 0
else
return 1
fi
}
base_dir=$(dirname $0)/..
if [ -z "$SCALA_VERSION" ]; then
SCALA_VERSION=2.13.5
if [[ -f "$base_dir/gradle.properties" ]]; then
SCALA_VERSION=`grep "^scalaVersion=" "$base_dir/gradle.properties" | cut -d= -f 2`
fi
fi
if [ -z "$SCALA_BINARY_VERSION" ]; then
SCALA_BINARY_VERSION=$(echo $SCALA_VERSION | cut -f 1-2 -d '.')
fi
# run ./gradlew copyDependantLibs to get all dependant jars in a local dir
shopt -s nullglob
if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
for dir in "$base_dir"/core/build/dependant-libs-${SCALA_VERSION}*;
do
CLASSPATH="$CLASSPATH:$dir/*"
done
fi
for file in "$base_dir"/examples/build/libs/kafka-examples*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
clients_lib_dir=$(dirname $0)/../clients/build/libs
streams_lib_dir=$(dirname $0)/../streams/build/libs
streams_dependant_clients_lib_dir=$(dirname $0)/../streams/build/dependant-libs-${SCALA_VERSION}
else
clients_lib_dir=/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs
streams_lib_dir=$clients_lib_dir
streams_dependant_clients_lib_dir=$streams_lib_dir
fi
for file in "$clients_lib_dir"/kafka-clients*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
for file in "$streams_lib_dir"/kafka-streams*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
for file in "$base_dir"/streams/examples/build/libs/kafka-streams-examples*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
else
VERSION_NO_DOTS=`echo $UPGRADE_KAFKA_STREAMS_TEST_VERSION | sed 's/\.//g'`
SHORT_VERSION_NO_DOTS=${VERSION_NO_DOTS:0:((${#VERSION_NO_DOTS} - 1))} # remove last char, ie, bug-fix number
for file in "$base_dir"/streams/upgrade-system-tests-$SHORT_VERSION_NO_DOTS/build/libs/kafka-streams-upgrade-system-tests*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$file":"$CLASSPATH"
fi
done
if [ "$SHORT_VERSION_NO_DOTS" = "0100" ]; then
CLASSPATH="/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs/zkclient-0.8.jar":"$CLASSPATH"
CLASSPATH="/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs/zookeeper-3.4.6.jar":"$CLASSPATH"
fi
if [ "$SHORT_VERSION_NO_DOTS" = "0101" ]; then
CLASSPATH="/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs/zkclient-0.9.jar":"$CLASSPATH"
CLASSPATH="/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs/zookeeper-3.4.8.jar":"$CLASSPATH"
fi
fi
for file in "$streams_dependant_clients_lib_dir"/rocksdb*.jar;
do
CLASSPATH="$CLASSPATH":"$file"
done
for file in "$streams_dependant_clients_lib_dir"/*hamcrest*.jar;
do
CLASSPATH="$CLASSPATH":"$file"
done
for file in "$base_dir"/shell/build/libs/kafka-shell*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
for dir in "$base_dir"/shell/build/dependant-libs-${SCALA_VERSION}*;
do
CLASSPATH="$CLASSPATH:$dir/*"
done
for file in "$base_dir"/tools/build/libs/kafka-tools*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
for dir in "$base_dir"/tools/build/dependant-libs-${SCALA_VERSION}*;
do
CLASSPATH="$CLASSPATH:$dir/*"
done
for cc_pkg in "api" "transforms" "runtime" "file" "mirror" "mirror-client" "json" "tools" "basic-auth-extension"
do
for file in "$base_dir"/connect/${cc_pkg}/build/libs/connect-${cc_pkg}*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
if [ -d "$base_dir/connect/${cc_pkg}/build/dependant-libs" ] ; then
CLASSPATH="$CLASSPATH:$base_dir/connect/${cc_pkg}/build/dependant-libs/*"
fi
done
# classpath addition for release
for file in "$base_dir"/libs/*;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
# CONFLUENT: classpath addition for releases with LSB-style layout
CLASSPATH="$CLASSPATH":"$base_dir/share/java/kafka/*"
# classpath for telemetry
CLASSPATH="$CLASSPATH":"$base_dir/share/java/confluent-telemetry/*"
for file in "$base_dir"/core/build/libs/kafka_${SCALA_BINARY_VERSION}*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
shopt -u nullglob
if [ -z "$CLASSPATH" ] ; then
echo "Classpath is empty. Please build the project first e.g. by running './gradlew jar -PscalaVersion=$SCALA_VERSION'"
exit 1
fi
# JMX settings
if [ -z "$KAFKA_JMX_OPTS" ]; then
KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false "
fi
# JMX port to use
if [ $JMX_PORT ]; then
KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
fi
# Log directory to use
if [ "x$LOG_DIR" = "x" ]; then
LOG_DIR="$base_dir/logs"
fi
# Log4j settings
if [ -z "$KAFKA_LOG4J_OPTS" ]; then
# Log to console. This is a tool.
LOG4J_CONFIG_NORMAL_INSTALL="/etc/kafka/tools-log4j.properties"
LOG4J_CONFIG_ZIP_INSTALL="$base_dir/etc/kafka/tools-log4j.properties"
if [ -e "$LOG4J_CONFIG_NORMAL_INSTALL" ]; then # Normal install layout
LOG4J_DIR="${LOG4J_CONFIG_NORMAL_INSTALL}"
elif [ -e "${LOG4J_CONFIG_ZIP_INSTALL}" ]; then # Simple zip file layout
LOG4J_DIR="${LOG4J_CONFIG_ZIP_INSTALL}"
else # Fallback to normal default
LOG4J_DIR="$base_dir/config/tools-log4j.properties"
fi
# If Cygwin is detected, LOG4J_DIR is converted to Windows format.
(( CYGWIN )) && LOG4J_DIR=$(cygpath --path --mixed "${LOG4J_DIR}")
KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_DIR}"
else
# create logs directory
if [ ! -d "$LOG_DIR" ]; then
mkdir -p "$LOG_DIR"
fi
fi
# If Cygwin is detected, LOG_DIR is converted to Windows format.
(( CYGWIN )) && LOG_DIR=$(cygpath --path --mixed "${LOG_DIR}")
KAFKA_LOG4J_OPTS="-Dkafka.logs.dir=$LOG_DIR $KAFKA_LOG4J_OPTS"
# Generic jvm settings you want to add
if [ -z "$KAFKA_OPTS" ]; then
KAFKA_OPTS=""
fi
# Set Debug options if enabled
if [ "x$KAFKA_DEBUG" != "x" ]; then
# Use default ports
DEFAULT_JAVA_DEBUG_PORT="5005"
if [ -z "$JAVA_DEBUG_PORT" ]; then
JAVA_DEBUG_PORT="$DEFAULT_JAVA_DEBUG_PORT"
fi
# Use the defaults if JAVA_DEBUG_OPTS was not set
DEFAULT_JAVA_DEBUG_OPTS="-agentlib:jdwp=transport=dt_socket,server=y,suspend=${DEBUG_SUSPEND_FLAG:-n},address=$JAVA_DEBUG_PORT"
if [ -z "$JAVA_DEBUG_OPTS" ]; then
JAVA_DEBUG_OPTS="$DEFAULT_JAVA_DEBUG_OPTS"
fi
echo "Enabling Java debug options: $JAVA_DEBUG_OPTS"
KAFKA_OPTS="$JAVA_DEBUG_OPTS $KAFKA_OPTS"
fi
# Which java to use
if [ -z "$JAVA_HOME" ]; then
JAVA="java"
else
JAVA="$JAVA_HOME/bin/java"
fi
# Memory options
if [ -z "$KAFKA_HEAP_OPTS" ]; then
KAFKA_HEAP_OPTS="-Xmx256M"
fi
# JVM performance options
# MaxInlineLevel=15 is the default since JDK 14 and can be removed once older JDKs are no longer supported
if [ -z "$KAFKA_JVM_PERFORMANCE_OPTS" ]; then
KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true"
fi
while [ $# -gt 0 ]; do
COMMAND=$1
case $COMMAND in
-name)
DAEMON_NAME=$2
CONSOLE_OUTPUT_FILE=$LOG_DIR/$DAEMON_NAME.out
shift 2
;;
-loggc)
if [ -z "$KAFKA_GC_LOG_OPTS" ]; then
GC_LOG_ENABLED="true"
fi
shift
;;
-daemon)
DAEMON_MODE="true"
shift
;;
*)
break
;;
esac
done
# GC options
GC_FILE_SUFFIX='-gc.log'
GC_LOG_FILE_NAME=''
if [ "x$GC_LOG_ENABLED" = "xtrue" ]; then
GC_LOG_FILE_NAME=$DAEMON_NAME$GC_FILE_SUFFIX
# The first segment of the version number, which is '1' for releases before Java 9
# it then becomes '9', '10', ...
# Some examples of the first line of `java --version`:
# 8 -> java version "1.8.0_152"
# 9.0.4 -> java version "9.0.4"
# 10 -> java version "10" 2018-03-20
# 10.0.1 -> java version "10.0.1" 2018-04-17
# We need to match to the end of the line to prevent sed from printing the characters that do not match
JAVA_MAJOR_VERSION=$("$JAVA" -version 2>&1 | sed -E -n 's/.* version "([0-9]*).*$/\1/p')
if [[ "$JAVA_MAJOR_VERSION" -ge "9" ]] ; then
KAFKA_GC_LOG_OPTS="-Xlog:gc*:file=$LOG_DIR/$GC_LOG_FILE_NAME:time,tags:filecount=10,filesize=100M"
else
KAFKA_GC_LOG_OPTS="-Xloggc:$LOG_DIR/$GC_LOG_FILE_NAME -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M"
fi
fi
# Remove a possible colon prefix from the classpath (happens at lines like `CLASSPATH="$CLASSPATH:$file"` when CLASSPATH is blank)
# Syntax used on the right side is native Bash string manipulation; for more details see
# http://tldp.org/LDP/abs/html/string-manipulation.html, specifically the section titled "Substring Removal"
CLASSPATH=${CLASSPATH#:}
# If Cygwin is detected, classpath is converted to Windows format.
(( CYGWIN )) && CLASSPATH=$(cygpath --path --mixed "${CLASSPATH}")
# Launch mode
if [ "x$DAEMON_MODE" = "xtrue" ]; then
nohup "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
exec "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@"
fi
/opt/confluent-6.2.0/bin/schema-registry-start
schema-registry-start
#!/bin/bash
#
# Copyright 2018 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
print_synopsis() {
echo "USAGE: $0 [-daemon] schema-registry.properties"
}
EXTRA_ARGS=${EXTRA_ARGS-'-name schemaRegistry'}
# EXTRA_ARGS=${EXTRA_ARGS-'-name schemaRegistry -loggc'}
ARG=$1
case $ARG in
-daemon)
EXTRA_ARGS="-daemon "$EXTRA_ARGS
shift
;;
*)
;;
esac
if [[ $# -lt 1 ]]; then
print_synopsis
exit 1
fi
if [[ ! -e "$1" ]]; then
echo "Property file $1 does not exist"
print_synopsis
exit 1
fi
exec $(dirname $0)/schema-registry-run-class ${EXTRA_ARGS} io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain "$@"
/opt/confluent-6.2.0/bin/schema-registry-stop
schema-registry-stop
#!/bin/bash
#
# Copyright 2018 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# When stopping, search for both the current SchemaRegistryMain class and the deprecated Main class.
exec $(dirname $0)/schema-registry-stop-service "(io.confluent.kafka.schemaregistry.rest.Main)|(io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain)"
/opt/confluent-6.2.0/bin/schema-registry-run-class
schema-registry-run-class
#!/bin/bash
#
# Copyright 2018 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
if [ $# -lt 1 ];
then
echo "USAGE: $0 [-daemon] [-name servicename] [-loggc] classname [opts]"
exit 1
fi
base_dir=$(dirname $0)/..
# CYGINW == 1 if Cygwin is detected, else 0.
if [[ $(uname -a) =~ "CYGWIN" ]]; then
CYGWIN=1
else
CYGWIN=0
fi
# Development jars. `mvn package` should collect all the required dependency jars here
for dir in $base_dir/package-schema-registry/target/kafka-schema-registry-package-*-development; do
CLASSPATH=$CLASSPATH:$dir/share/java/schema-registry/*
done
# Production jars, including kafka, rest-utils, and schema-registry
for library in "confluent-security/schema-registry" "confluent-common" "confluent-telemetry" "rest-utils" "schema-registry"; do
CLASSPATH=$CLASSPATH:$base_dir/share/java/$library/*
done
# Log directory to use
if [ "x$LOG_DIR" = "x" ]; then
LOG_DIR="$base_dir/logs"
fi
# create logs directory
if [ ! -d "$LOG_DIR" ]; then
mkdir -p "$LOG_DIR"
fi
# logj4 settings
if [ "x$SCHEMA_REGISTRY_LOG4J_OPTS" = "x" ]; then
# Test for files from dev -> packages so this will work as expected in dev if you have packages
# installed
if [ -e "$base_dir/config/log4j.properties" ]; then # Dev environment
LOG4J_DIR="$base_dir/config/log4j.properties"
elif [ -e "$base_dir/etc/schema-registry/log4j.properties" ]; then # Simple zip file layout
LOG4J_DIR="$base_dir/etc/schema-registry/log4j.properties"
elif [ -e "/etc/schema-registry/log4j.properties" ]; then # Normal install layout
LOG4J_DIR="/etc/schema-registry/log4j.properties"
fi
# If Cygwin is detected, LOG4J_DIR is converted to Windows format.
(( CYGWIN )) && LOG4J_DIR=$(cygpath --path --mixed "${LOG4J_DIR}")
SCHEMA_REGISTRY_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_DIR}"
fi
# If Cygwin is detected, LOG_DIR is converted to Windows format.
(( CYGWIN )) && LOG_DIR=$(cygpath --path --mixed "${LOG_DIR}")
SCHEMA_REGISTRY_LOG4J_OPTS="-Dschema-registry.log.dir=$LOG_DIR $SCHEMA_REGISTRY_LOG4J_OPTS"
# JMX settings
if [ -z "$SCHEMA_REGISTRY_JMX_OPTS" ]; then
SCHEMA_REGISTRY_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false "
fi
# JMX port to use
if [ $JMX_PORT ]; then
SCHEMA_REGISTRY_JMX_OPTS="$SCHEMA_REGISTRY_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
fi
# Generic jvm settings you want to add
if [ -z "$SCHEMA_REGISTRY_OPTS" ]; then
SCHEMA_REGISTRY_OPTS=""
fi
# Which java to use
if [ -z "$JAVA_HOME" ]; then
JAVA="java"
else
JAVA="$JAVA_HOME/bin/java"
fi
# Memory options
if [ -z "$SCHEMA_REGISTRY_HEAP_OPTS" ]; then
SCHEMA_REGISTRY_HEAP_OPTS="-Xmx512M"
fi
# JVM performance options
if [ -z "$SCHEMA_REGISTRY_JVM_PERFORMANCE_OPTS" ]; then
SCHEMA_REGISTRY_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true"
fi
while [ $# -gt 0 ]; do
COMMAND=$1
case $COMMAND in
-help)
HELP="true"
break
;;
-name)
DAEMON_NAME=$2
CONSOLE_OUTPUT_FILE=$LOG_DIR/$DAEMON_NAME.out
shift 2
;;
-loggc)
if [ -z "$SCHEMA_REGISTRY_GC_LOG_OPTS" ]; then
GC_LOG_ENABLED="true"
fi
shift
;;
-daemon)
DAEMON_MODE="true"
shift
;;
*)
break
;;
esac
done
if [ "x$$HELP" = "xtrue" ]; then
echo "USAGE: $0 [-daemon] [-name servicename] [-loggc] classname [opts]"
exit 0
fi
MAIN=$1
shift
# GC options
GC_FILE_SUFFIX='-gc.log'
GC_LOG_FILE_NAME=''
if [ "x$GC_LOG_ENABLED" = "xtrue" ]; then
GC_LOG_FILE_NAME=$DAEMON_NAME$GC_FILE_SUFFIX
# The first segment of the version number, which is '1' for releases before Java 9
# it then becomes '9', '10', ...
# Some examples of the first line of `java --version`:
# 8 -> java version "1.8.0_152"
# 9.0.4 -> java version "9.0.4"
# 10 -> java version "10" 2018-03-20
# 10.0.1 -> java version "10.0.1" 2018-04-17
# We need to match to the end of the line to prevent sed from printing the characters that do not match
JAVA_MAJOR_VERSION=$($JAVA -version 2>&1 | sed -E -n 's/.* version "([0-9]*).*$/\1/p')
if [[ "$JAVA_MAJOR_VERSION" -ge "9" ]] ; then
SCHEMA_REGISTRY_GC_LOG_OPTS="-Xlog:gc*:file=$LOG_DIR/$GC_LOG_FILE_NAME:time,tags:filecount=10,filesize=102400"
else
SCHEMA_REGISTRY_GC_LOG_OPTS="-Xloggc:$LOG_DIR/$GC_LOG_FILE_NAME -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M"
fi
fi
# If Cygwin is detected, classpath is converted to Windows format.
(( CYGWIN )) && CLASSPATH=$(cygpath --path --mixed "${CLASSPATH}")
# Launch mode
if [ "x$DAEMON_MODE" = "xtrue" ]; then
CONSOLE_OUTPUT_FILE=${CONSOLE_OUTPUT_FILE:-${LOG_DIR}/schema-registry-console.out}
nohup $JAVA $SCHEMA_REGISTRY_HEAP_OPTS $SCHEMA_REGISTRY_JVM_PERFORMANCE_OPTS $SCHEMA_REGISTRY_GC_LOG_OPTS $SCHEMA_REGISTRY_JMX_OPTS $SCHEMA_REGISTRY_LOG4J_OPTS -cp $CLASSPATH $SCHEMA_REGISTRY_OPTS "$MAIN" "$@" > "${CONSOLE_OUTPUT_FILE}" 2>&1 < /dev/null &
else
exec "$JAVA" $SCHEMA_REGISTRY_HEAP_OPTS $SCHEMA_REGISTRY_JVM_PERFORMANCE_OPTS $SCHEMA_REGISTRY_GC_LOG_OPTS $SCHEMA_REGISTRY_JMX_OPTS $SCHEMA_REGISTRY_LOG4J_OPTS -cp $CLASSPATH $SCHEMA_REGISTRY_OPTS "$MAIN" "$@"
fi