メモ：Windows上でapache kafka＋kafka-node



分散メッセージキューkafkaそのものについては[ここ](https://qiita.com/sigmalist/items/5a26ab519cbdf1e07af3)の解説が詳しい。
あとは[これ](https://qiita.com/YutaSaito1991/items/214ee6d5a81ac4436cd4)とか。
AWSのAmazonMSKのようなフルマネージドが楽そうだが、まずはWindows上でどんな感じか試してみる。
環境は少し古いWindows Server 2012
node.jsは16.x

# kafkaインストール
[ここ](https://towardsdatascience.com/running-zookeeper-kafka-on-windows-10-14fc70dcc771)とか[ここ](http://java-study.blog.jp/archives/1075373530.html)を見ながら。

* とりあえず[https://kafka.apache.org/downloads](https://kafka.apache.org/downloads)からバイナリダウンロード
* 7zipで解凍して適当なフォルダに配置
* zookeeperは組み込みを利用するので、`bin\windows\zookeeper-server-start.bat config\zookeeper.properties`で起動
* 別のプロンプトを立ち上げて`bin\windows\kafka-server-start.bat config/server.properties`

起動してみると`java.nio.file.AccessDeniedException`のエラー
config下にあるlog.dirs関係を全て`log.dirs=G:\tmp\kraft-combined-logs`のようにWindows形式のパスに変更すると
`java.nio.file.InvalidPathException: Illegal char <      > at index 2: G:        mpkafka-logs\meta.properties.tmp`
となったので、\を\\でエスケープしてみるがやはりAccessDeniedExceptionエラー。
フォルダの権限の問題でもない。
調べてみると単純に2.12-3.0.0の問題のようなので古いバージョンのkafka_2.12-2.8.1に変更すると設定を変更しなくても動いた。

さらに別のコンソールを開いてトピックを作成して確認
topc名(test)が出てくるはず。
partitionsはとりあえず1で試せばよいが、consumerが複数の場合はそれ以上のpartitionsが存在している必要あり。


```
bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
```

なお、削除は
`bin\windows\kafka-topics.bat --delete --zookeeper localhost:2181 --topic test`


# メッセージ送受信（kafka-node）
[ダウンロード数](https://www.npmtrends.com/kafkajs-vs-kafka-node-vs-node-rdkafka)を見るとkafkajsやnode-rdkafkaよりkafuka-nodの方が優勢で、
サンプルも多そうなのでkafka-nodeを利用する。


[ここ](https://dev.classmethod.jp/articles/kafka-streams/)とかでnode.jsからのアクセス方法が書いてある。

`npm install kafka-node`

サンプルとは少し変えて、expressでGETで投げたものを適当に直接kafkaに投げてみる

```javascript
'use strict';
const kafka = require("kafka-node");
const express = require('express');
const app = express();
const port = 3000;
const Producer = kafka.HighLevelProducer;
const Consumer = kafka.Consumer;
const client = new kafka.KafkaClient({
    kafkaHost: "localhost:9092"
});
const producer = new Producer(client, {
	// Partitioner type (default = 0, random = 1, cyclic = 2, keyed = 3, custom = 4), default 0
    partitionerType: 1
});
const consumer = new Consumer(
    client,
    [{topic: "test", partision:0}],
    {
        groupId: "my-consumer",
        autoCommit: true,
        fromOffset: true
    }
);

app.get("/", function(req, res, next){
    const name = req.query.name;
    const age = req.query.age;
    const message = [
        {
            topic: "test",
            messages: JSON.stringify({name: name, age: age})
        }
    ];
    producer.send(message, (err, data) => {
        if (err) console.log(err);
        else console.log('send messages');
        //process.exit();
    });
    res.status(200).send("SEND"); // テストなので常に成功扱い
});

// 受信する場合
consumer.on("message", (message, err) => {
    console.log(message);
});

app.listen(port, () => console.log(`Example app listening on port ${port}!`));
```

送信出来ているかどうかをコマンドでも確認してみると、飛んできている事がわかる。
`--from-beginning`を指定すると最初から読み込む。


```
bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
```

受信にはいくつかパターンがあるようで、[ここ](https://qiita.com/proyuki02/items/cc30f0248315aaf3573e)見ると
`new kafka.ConsumerGroupStream`を使うのが一番高機能っぽい。
詳細は[公式](https://www.npmjs.com/package/kafka-node)


SQLを使ってkafkaをコントロールする[KSQL](https://qiita.com/cohhei/items/b897137be06f3f1fac26)のようなものもあるがMSKは対応していなさそう。

# マルチConsumer化
起動時に
`Error while executing topic command : Replication factor: 1 larger than available brokers: 0.`
というようなエラーが出て立ち上がらない事があったが、tmpを全て削除して、アプリも終了させてから起動させると立ち上がった。

tmp等を削除してから再起動して
`bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic test`
としてpartitionを複数で立ち上げる。

Consumerは先のnode.jsのものからポートを3001に変更したものを作成。
partitionを指定しているため、`[{topic: "test", partition:1}],`のようにして別のものを指定して起動。
ブラウザから登録してみると別々の受信をしている事が分かる。

とはいえ、これでは自動でpartitionが割り振られないので使いにくい。
ConsumerGroupStreamを使ってグループで割り振られるようにする。
２つ用意するが、ポートの違いのみ。

```javascript
'use strict';
const kafka = require("kafka-node");
const express = require('express');
const app = express();
const port = 3000;
const Producer = kafka.HighLevelProducer;

const producer = new Producer(client, {
    partitionerType: 1
});

const consumer = new kafka.ConsumerGroupStream({
    kafkaHost: "localhost:9092",
    groupId: 'my-consumer',
    autoCommit: true,
    fromOffset: 'earliest'
  }, 'test')

app.get("/send", function(req, res, next){
    const name = req.query.name;
    const age = req.query.age;
    const message = [
        {
            topic: "test",
            messages: JSON.stringify({name: name, age: age})
        }
    ];
    producer.send(message, (err, data) => {
        if (err) {console.log(err); res.status(500);}
        else console.log('send messages');
        //process.exit();
    });
    res.status(200).send("SEND");
});

// messageではなくdataになる
consumer.on("data", (message, err) => {
    console.log(message);
});

app.listen(port, () => console.log(`Example app listening on port ${port}!`));
```

これで実験すると各々が異なるデータを受信出来たので概ねOKぽい。