[Oracle Cloud] Streaming に Kafka Connect を使って接続してみた

# はじめに

Oracle Cloud Infractructure(以下OCI)では、ストリーミングデータをリアルタイムに収集・処理が出来る Streaming というサービスが提供されています。Streaming へデータを入出力する方法は、ざっくり次のものがあります

- OCI SDK
- Kafka 互換API
- Kafka Connect
- OCI CLI
- REST API



今回の記事では、Kafka Connect を使って Streaming にデータを入出力してみます。Kafka Connect は、Apache Kafka で提供されている API となっていて、簡単にデータを入出力出来ます。Connect に接続するための Connector が色々公開されているため、自分が使いたい Connector があれば、自分で実装する負担を減らせます。使いたい Connector が無ければ、自分で作っていくこともできます。

インターネット上に公開されているコネクターは、ライセンス形態にご注意ください。



# Auth Token生成

Streaming で、Kafka Connect で接続するためには、IAM ユーザーで Auth Token が必要です。自分の IAMユーザーの詳細画面で、Generate します。

![1588474181611.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/d5e213cd-9d8d-61ca-51d9-d09a52f3bf37.png)



適当に説明をいれます。

![1588474211704.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/826783fd-91d3-695a-fdf6-49292d958f01.png)



Token が表示されるので、メモっておきます。(画像のものは現在使えません)

**Tokenの中に、; などの記号が含まれていると、正しく動かない気がします。もしハマって正常に動かない時は、Tokenの入れ替えも検討してみてください。**

![1588474230416.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/b181dd69-3e36-ce09-c95e-3c6d1dd97f76.png)



# Stream の作成

OCI Console で、Analytics > Streaming を選択します。その後、Create Stream を選びます。

![1588519801919.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/dedbe86d-6667-2f82-6f98-a18acbcc18f5.png)



適当にパラメータを入れて作成します

![1588519908911.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/a12f4d72-c6bd-b10d-533e-d52d94d7a12d.png)



Stream を作ったときに、Stream Pool が何もない場合は、Default の Stream Pool が自動的に作成されます。`teststream01`は、DefaultPool に自動配置されます。Stream Pool は、複数のStreamを管理する概念となっていて、Stream への Endpoint を Public にするのか Private にするのか、またデータの暗号に使う鍵は何を使用するのか、といった事をまとめて管理するための概念です。

今回は自動作成にしたので、Endpoint は Public になっていて、暗号化用の鍵は Oracle 側で管理されているものを使う設定です。

![1588520017422.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/d02f1ff7-aaed-1f31-b943-1ef534b77d7c.png)



上記の画面にある「View Kafka Connection Settings」を押します。

![1588521024252.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/9e75dd0e-7ce9-71f9-7962-fb86d32dacdf.png)



Copy All を押して、すべての設定値をメモっておきます。

![1588521056591.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/c932646e-6fa7-f74f-8f74-59e40736f822.png)



# Kafka Connect Configration を作成

Kafka Connector を使ってStreamingに接続するために、Kafka Connect Configration を作成します。これを作成すると、Streaming 側で Kafka 用語でいうところの  3種類のTopic が作成されます。複数の Connector が負荷分散して動作するためには、互いにどういうステータスで、どこまでのデータを処理して、といった情報共有が必要です。情報共有するために、3種類のTopic上でやり取りを行っています。

それでは、Create ボタンをおします。

![1588521439761.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/80c0eb5f-7522-f202-4edf-620ffb820793.png)



名前を適当にいれます。

![1588521472264.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/72e13e28-ea86-a555-9d1b-12f437ae3514.png)



詳細画面が自動的に表示されます。Copy を押してメモっておきます。

![1588521505413.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/934cd26f-5ff1-9612-a542-395250036434.png)







# Connector 用仮想マシンの作成

今回の手順では、適当に CentOS7 の仮想マシンを作成します。



# Open JDK 8 の Install

作成した CentOS7 で Kafka Connector を動かすために、Open JDK 1.8 を Install します。好きな JDK で問題ないと思います。

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



# Kafka Download

Kafka Connector を動かすためのツールが、Kafka 本体の tar gz ball に含まれているのでダウンロードします。以下の URL からダウンロードできます。

https://kafka.apache.org/quickstart

```
mkdir ~/kafka
cd ~/kafka
wget https://ftp.tsukuba.wide.ad.jp/software/apache/kafka/2.5.0/kafka_2.12-2.5.0.tgz
```



解凍

```
tar xfvz kafka_2.12-2.5.0.tgz
```



作業しやすいために、Kafka Home を環境変数に設定

```
echo 'export KAFKA_HOME=$HOME/kafka/kafka_2.12-2.5.0/' >> ~/.bashrc
```



bachrc 読み込み

```
source ~/.bashrc
```



Kafka の targz ball に含まれているライブラリを確認します。

```
ls -la $KAFKA_HOME/libs | grep -i connect
```



いくつかの connector が Kafka に含まれています。

```
[opc@kafka01 libs]$ ls -la $KAFKA_HOME/libs | grep -i connect
-rw-r--r--. 1 opc opc   101941 Apr  8 01:14 connect-api-2.5.0.jar
-rw-r--r--. 1 opc opc    18607 Apr  8 01:16 connect-basic-auth-extension-2.5.0.jar
-rw-r--r--. 1 opc opc    20506 Apr  8 01:16 connect-file-2.5.0.jar
-rw-r--r--. 1 opc opc    45199 Apr  8 01:14 connect-json-2.5.0.jar
-rw-r--r--. 1 opc opc    90298 Apr  8 01:16 connect-mirror-2.5.0.jar
-rw-r--r--. 1 opc opc    29086 Apr  8 01:16 connect-mirror-client-2.5.0.jar
-rw-r--r--. 1 opc opc   557008 Apr  8 01:16 connect-runtime-2.5.0.jar
-rw-r--r--. 1 opc opc    91639 Apr  8 01:16 connect-transforms-2.5.0.jar
[opc@kafka01 libs]$
```



# Kafka Connect 用設定

Kafka Connector 用に必要な設定ファイルを作成します。それぞれのパラメータは、環境に合わせて適宜変更してください。

- `group.id` : 任意の名前。複数の Connect Worker を動かす時は、どういうの名前を指定する
- `bootstrap.servers` : OCI の Stream Pool からコピーした値を指定
- `sasl.jaas.config` : OCI の Stream Pool からコピーした値を指定。`password=""` の部分は、Auth Token の文字列を指定
- `producer.sasl.jaas.config` : `sasl.jaas.config` と同じものを指定
- `consumer.sasl.jaas.config` : `sasl.jaas.config` と同じものを指定
- `config.storage.topic` : Kafka Connect Configration からコピーした値を指定
- `status.storage.topic` : Kafka Connect Configration からコピーした値を指定
- `offset.storage.topic` : Kafka Connect Configration からコピーした値を指定

```
cat <<'EOF' > $KAFKA_HOME/config/connect-distributed-demo.properties
group.id=connect-demo-group

bootstrap.servers=cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla" password="8t[shwUN}I-d+{}8Nx_a";

producer.sasl.mechanism=PLAIN
producer.security.protocol=SASL_SSL
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla" password="8t[shwUN}I-d+{}8Nx_a";

consumer.sasl.mechanism=PLAIN
consumer.security.protocol=SASL_SSL
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla" password="8t[shwUN}I-d+{}8Nx_a";

config.storage.replication.factor=1
config.storage.partitions=1
config.storage.topic=ocid1.connectharness.oc1.ap-tokyo-1.amaaaaaaycetm7yapmslajk3scdwvngqw47c2iixkqltnb7zllk7ca3t33va-config

offset.storage.replication.factor=1
offset.storage.partitions=1
offset.storage.topic=ocid1.connectharness.oc1.ap-tokyo-1.amaaaaaaycetm7yapmslajk3scdwvngqw47c2iixkqltnb7zllk7ca3t33va-offset
offset.flush.interval.ms=10000
offset.flush.timeout.ms=5000

status.storage.replication.factor=1
status.storage.partitions=1
status.storage.topic=ocid1.connectharness.oc1.ap-tokyo-1.amaaaaaaycetm7yapmslajk3scdwvngqw47c2iixkqltnb7zllk7ca3t33va-status

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=true

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter

task.shutdown.graceful.timeout.ms=10000
EOF
```



Kafka Connect を実行

```
$KAFKA_HOME/bin/connect-distributed.sh $KAFKA_HOME/config/connect-distributed-demo.properties
```



実行例
たくさんの文字が流れて行って、末尾にこんな文字が表示されれば正常状態です。

```
[2020-05-03 03:07:46,369] INFO [Worker clientId=connect-1, groupId=connect-demo-group] Finished reading to end of log and updated config snapshot, new config log offset: 886603188224 (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1085)[2020-05-03 03:07:46,369] INFO [Worker clientId=connect-1, groupId=connect-demo-group] Starting connectors and tasks using config offset 886603188224 (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1111)
[2020-05-03 03:07:46,369] INFO [Worker clientId=connect-1, groupId=connect-demo-group] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1132)[2020-05-03 03:07:47,378] INFO [Worker clientId=connect-1, groupId=connect-demo-group] Session key updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1447)
```



# Source Connector の設定 

今回の記事では、File Connector を使用して動作確認をします。動作確認に使う Inputファイル・Outputファイルを作成します。Input ファイルは、1から1000までの文字を1行ごとにかいたファイルです。output.txt は touch で作っているだけなので、中身は空っぽです。

```
mkdir ~/connect-quickstart
seq 1000 > ~/connect-quickstart/input.txt
touch ~/connect-quickstart/output.txt
```



コネクタを作成するまえに、Connect Worker(仮想マシン) の中で定義されている Connector 一覧を表示します。今はなにもしていないので、空白が返ってきます。

```
[opc@kafka01 connect-quickstart]$ curl -s http://localhost:8083/connectors | jq .
[]
[opc@kafka01 connect-quickstart]$
```



Connector を設定します。

- `topic` : OCI Stream の名前を指定
- `file` : Input として使うファイルのPathを指定

```
curl -s -X POST -H "Content-Type: application/json" --data '{"name": "file-source","config": {"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector","tasks.max":"1","topic":"teststream01","file": "/home/opc/connect-quickstart/input.txt"}}' http://localhost:8083/connectors
```



list を確認します。

```
[opc@kafka01 connect-quickstart]$ curl -s http://localhost:8083/connectors | jq .
[
  "file-source"
]
[opc@kafka01 connect-quickstart]$
```



詳細情報を確認します

```
[opc@kafka01 connect-quickstart]$ curl -s http://localhost:8083/connectors/file-source/status | jq .
{
  "name": "file-source",
  "connector": {
    "state": "RUNNING",
    "worker_id": "10.0.0.14:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "10.0.0.14:8083"
    }
  ],
  "type": "source"
}
```



memo : Deleteコマンド

```
curl -s -X DELETE http://localhost:8083/connectors/file-source | jq .
```



# Sink Connector の設定

Connector の Output(Sink) としての設定をします。こちらも Input と同様に File Connector を使います。

```
curl -X POST -H "Content-Type: application/json" --data '{"name": "file-sink", "config": {"connector.class":"org.apache.kafka.connect.file.FileStreamSinkConnector", "tasks.max":"1", "topics":"teststream01", "file": "/home/opc/connect-quickstart/output.txt"}}' http://localhost:8083/connectors
```



fileが作成されています。

```
[opc@kafka01 connect-quickstart]$ head -n 10 output.txt
1
2
3
4
5
6
7
8
9
10
[opc@kafka01 connect-quickstart]$
```





# 検証してわかったこと

- Streaming の Stream の名前が、KafkaのTopicNameに対応している
- 一度コネクターを作成した後に、input file を更新しても output には流れない





# 参考URL

Kafka Docs
https://kafka.apache.org/documentation/

OCI Docs
https://docs.cloud.oracle.com/en-us/iaas/Content/Streaming/Tasks/kafkacompatibility.htm

Oracle Blogs
https://blogs.oracle.com/developers/using-kafka-connect-with-oracle-streaming-service-and-autonomous-db