Apache Kafkaを試してみる

#Apache Kafka
Apache Kafkaが必要になりそうなので、動作確認してみました。
Kafkaとは分散メッセージングプラットフォームで、メッセージをやり取りするプラットフォームです。
##既存メッセージングとの違い
JMSなど既存のメッセージングとは下記の点が異なります。

• 分散処理が前提になっており、耐障害性が高い&スケーリング可能
• メッセージの履歴を保持する事が可能で、ストレージ的な側面も持つ。
• ストリーム処理をすることが可能。入力メッセージに対して、様々な処理がリアルタイムにできる。

インストール

テスト用インストールは非常に簡単。
今回はAWS上のAmazon Linux上で作業しています。

Javaのインストール

まっさらな環境なのでJavaのインストールから始めます。

su -
yum install java-1.8.0-openjdk-devel

Kafkaのダウンロードと展開

この記事を書いている段階で1.1.0が最新です。

wget http://ftp.tsukuba.wide.ad.jp/software/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz
tar -zxf  kafka_2.11-1.1.0.tgz
cd kafka_2.11-1.1.0

これで準備は完了。

Kafkaの起動

Zookeeperは一言で言うと分散処理システムを構築するためのミドルウェアです。
Kafkaはzookeeperなしでも動作するようだけど、後ほどマルチブローカーのテストもするので使用します。

###Zookeeperを起動。

bin/zookeeper-server-start.sh config/zookeeper.properties
[2018-06-20 02:06:32,944] INFO Reading configuration from: config/zookeeper.properties 
(org.apache.zookeeper.server.quorum.QuorumPeerConfig)

ブローカーの起動

bin/kafka-server-start.sh config/server.properties
[2018-06-20 02:08:00,771] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2018-06-20 02:08:01,131] INFO starting (kafka.server.KafkaServer)

これでサービスとしては利用可能な状態

##通信テスト
###トピックの作成
試しにtestという名前のトピックを作成します。
トピックはメッセージの住所みたいなもので、トピックに向けてメッセージをやり取りします。

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

###作成されたトピックの確認

bin/kafka-topics.sh --list --zookeeper localhost:2181
test

###testトピックにメッセージを送信

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>hi this is from aws
>I'll send message

###コンシューマーからメッセージを確認

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
hi this is from aws
I'll send message

ストアされている既存のメッセージを取得できるのが新鮮！！

これでTOPICを作成し、メッセージのサブスクライブ、コンシューマーによるメッセージの確認という一連の流れは確認できた。

#マルチブローカーのテスト
Kafkaの特徴でもあるマルチブローカーをテストしてみたい。
##マルチブローカー設定
新規ブローカーようにコンフィグファイルをコピー
今回は3台構成にするので新たに2つのファイルをコピーする

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

##ファイルの編集
それぞれブローカーIDやリスナーポート、ログファイルの設定を変更。

config/server-1.propertiesの編集

vi config/server-1.properties
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9093
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs-1

###config/server-2.propertiesの編集

vi config/server-2.properties
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=2
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9094
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs-2

ブローカー起動

最初に設定したものと今回新たに設定した3台のブローカーを起動

bin/kafka-server-start.sh config/server.properties &
bin/kafka-server-start.sh config/server-1.properties &
bin/kafka-server-start.sh config/server-2.properties &

###レプリケーションファクタが3の新規トピックを作成
3つのブローカーに跨るトピックを作成してみる

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

###クラスタ状態の確認
新規に作成したトピック情報を取得

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 0       Replicas: 2,1,0 Isr: 1,0,2

最初の行はすべてのパーティションの要約で、それ以降の各行はパーティションに関する情報を示す。

• "Leader"は、パーティションのすべての読み取りと書き込みを担当するノードです。今回は0がリーダーになっている。
• "Replicas" はこのパーティションのログを複製するノードのリスト。
• "Isr"は "同期している"レプリカのセット

###既存TOPICとの比較
最初に作成したシングルブローカのトピックでは全てノード0が担当している。

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test      PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: test     Partition: 0    Leader: 0       Replicas: 0     Isr: 0

##クラスタ化されたトピックにメッセージを送信
###メッセージ送信

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
>this is clusterd Topic!
>next message

Consumerから確認

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
this is clusterd Topic!
next message

フォールトトレラントの確認

試しにリーダー、ノード0を落としてみます。

ps -ef | grep server 
ec2-user  9072  2984  2 03:15 pts/0 kafka.Kafka config/server.properties
kill -9 9072

zookeeper のログ

[2018-06-20 03:20:51,000] INFO Processed session termination for sessionid: 0x1641b2fd2520000 (org.apache.zookeeper.server.PrepRequestProcessor)

[1]   Killed                  bin/kafka-server-start.sh config/server.properties

再度クラスタの確認

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 2       Replicas: 2,1,0 Isr: 1,2

リーダーノード0が落ちたことにより自動的にノード2がリーダーになっていることが確認できた。