kafkaと戯れ

1machineでkafka環境を作って戯れる

## 1.環境設定

### OS＆JDK
CentOS7 をインストールしたのち

- firewalld をオフ
- selinux をオフ
- JDK8 をインストール

```
yum -y install java-1.8.0-openjdk java-1.8.0-openjdk-devel
```

- /etc/profile or ~/.bashrc にJAVA_HOME設定

```
export JAVA_HOME=/usr/lib/jvm/java
```

### ZooKeeper
CDHのRPMを利用する `/etc/yum.repos.d/` に以下を配置

```
https://archive.cloudera.com/cdh5/redhat/7/x86_64/cdh/cloudera-cdh5.repo
```
でyum install して、サービス起動と、自動起動設定もしておく

```
# yum -y install zookeeper-server
# service zookeeper-server start
# systemctl enable zookeeper-server.service
```

ポート2181が空いてれば起動成功かな `zookeeper-client` を使って確認してもよい

### Kafka
Apache サイトのバイナリtarball をダウンロード、展開

開発用なのでヒープサイズはちっちゃくてよい(デフォルトは1GB)

```
export KAFKA_HEAP_OPTS="-Xmx256M"
```

それに合わせて `config/server.properties`の各種値も1ケタ小さくしておこう

```
socket.request.max.bytes=10485760
log.retention.bytes=107374182
log.segment.bytes=107374182
```

そしてスタート

```
$ bin/kafka-server-start.sh config/server.properties &
```

疎通確認もしておく. topic 作って確認

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

メッセージを入れて取り出す 二つターミナルを用意して以下のようにする

```
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
```

うまくメッセージパッシングできていることを確認

## 2.Kafka Streamsとの戯れ

### デモの実行
v0.10で導入された kafka streams を試してみる。サンプルのデモを実行するため、二つのトピック `streams-file-input` と `streams-wordcount-output` を作成しておく

```
$ bin/kafka-topics.sh --create --zookeeper localhost:2181\
  --replication-factor 1 --partitions 1 --topic streams-file-input
$ bin/kafka-topics.sh --create --zookeeper localhost:2181\
  --replication-factor 1 --partitions 1 --topic streams-wordcount-output
```

input となるstreamの方には適当に、Apache Licenceの文章でも入れておく

```
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-file-input
**** Apache Licence の文章******
Ctrl+D
```

そして、クラスパスを通して、Streamアプリを実行

```
$  ./bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo
```

そして結果を確認

```
$ ./bin/kafka-console-consumer.sh \
--zookeeper localhost:2181 --topic streams-wordcount-output \
--from-beginning --formatter kafka.tools.DefaultMessageFormatter \
--property print.key=true \
--property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
--property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
```

### Kafka Streaming の概要

というか Kafka Streaming とはなんぞや？ドキュメント http://docs.confluent.io/2.1.0-alpha1/streams/index.html に概要が書かれているが、要は

* Kafka のクライアントライブラリとして実装された
* ストリーミングを実現するAPI群で
* トピックAのメッセージを処理してトピックBに詰め込むことを可能にする
* High Level なDSLと Low Level なインターフェイスを持ち
* スケーラブルでエラーセーフにすることもできる

とのこと

## 3.Kafka Streamsのアプリ開発

### 環境整備
男は黙ってvi. 以下のようなactivateファイルを用意して開発時には `source activate` とする

```bash
PS1=(kafka)$PS1
KAFKA_HOME=/home/______/kafka/kafka
for i in $KAFKA_HOME/libs/*.jar
do
CLASSPATH=$i:$CLASSPATH
done
export CLASSPATH
```

何回も実行して出力先のトピックが汚れるのもいやなので、毎回作り直す。以下のようなスクリプトを`init.sh`というように名前をつけて都度リフレッシュ。

```bash
./kafka/bin/kafka-topics.sh --delete --topic b --zookeeper localhost:2181
./kafka/bin/kafka-topics.sh --create --topic b --partitions 1 --replication-factor 1 --zookeeper localhost:2181
```

なお、トピックの削除には、サーバ側の設定が必要。以下のように設定して再起動。

```config/server.properties
delete.topic.enable=true
```

### テストコードのインターフェース

以下の「はまった点」にもあるように、同じ名前のアプリケーション名を使いまわしていると、前回の状態をKafka側が覚えている、すなわち同じコンシューマグループのクライアントとして認識されるため、試験をするときはアプリケーション名を都度変えて実行する





### はまった点

**さっきはうまくいっていたのに今回はうまくいかない**
:同じアプリケーション名だと、前回の状態を引き継ぐので、意図通りにConsumeできていない可能性がある