kafka-nodeのテスト

##### はじめに
マイクロサービスにおけるサービス間連携でnodejs × kafkaの利用シーンは増えています。
というわけでnodejs × kafkaの概念実証（PoC）メモ。
あくまでPoCなんで超簡易的です。

##### ディレクトリ構成
kafka/
　├ docker-compose.yml
　├ node_modules
　├ mackage.json
　└ test
　│　└ kafka-test.js
　└ src
　│　└ consumer
　│      └ app.js
　│　└ producer
　│      └ app.js  

| ファイル            |  説明    |実行コマンド|
|:-------------------|:------------------|:----|
| test/kafka-test.js |   Unit Test        |ava -v -u|
| consumer/app.js    |   Consumer実行ファイル |node src/consumer/app.js|
| producer/app.js    |   Producer実行ファイル |node src/producer/app.js|




#### セッティング
- docker-compose.yml

```
version: "2"

services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    links:
      - zookeeper
```

- kafkaトピック作成


```
$ docker-compose exec kafka /opt/kafka/bin/kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic test
$ docker-compose exec kafka /opt/kafka/bin/kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic test1
$ docker-compose exec kafka /opt/kafka/bin/kafka-topics.sh --list --zookeeper zookeeper:2181
# トピックからメッセージ取得
$ docker-compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic test1 --from-beginning
```

- package.json

```
{
  "name": "kafka",
  "version": "1.0.0",
  "description": "kafka-node",
  "main": "index.js",
  "dependencies": {
    "ava": "^1.4.1",
    "chai": "^4.2.0",
    "kafka-node": "^4.1.0"
  },
  "devDependencies": {},
  "scripts": {
    "test": "ava -v -u"
  },
  "author": "",
  "license": "ISC"
}
```

- test/kafka-test.js

```
import test from "ava";
import chai from "chai";
import kafka from "kafka-node";

// producerテスト
test.serial.cb("producer test", t => {
    const Producer = kafka.Producer;
    const client = new kafka.KafkaClient({
        kafkaHost: "192.168.33.50:9092"
    });
    const producer = new Producer(client, {
        partitionerType: 1
    });

    producer.on("ready", () => {
        const payloads = [
            {
                topic: "test1",
                messages: JSON.stringify({name: "神崎・H・アリア", age: 16})
            }
            ,{
                topic: "test",
                messages: [
                    JSON.stringify({name: "間宮あかり", age: 15}),
                   JSON.stringify({name: "佐々木志乃", age: 15})
               ]
            }
        ];

        producer.send(payloads, (err, data) => {
            t.end();
        });
    });
});

// consumerテスト
test.serial.cb("consumer test", t => {
    const Consumer = kafka.Consumer;
    const client = new kafka.KafkaClient({
        kafkaHost: "192.168.33.50:9092"
    });
    const consumer = new Consumer(
        client,
        [{topic: "test1", partision:0}],
        {
            groupId: "simple-consumer1",
            autoCommit: true,
            fromOffset: true
        }
    );

    consumer.on("message", (message, err) => {
        const json = JSON.stringify(message.value);
            chai.assert.isString(json.name);
            chai.assert.isNumber(json.age);
            if (json.age === 16) {
               chai.assert.propertyVal(json.attr,'name','神崎・H・アリア');
            }
    });
    t.end();
});

```

#### Unit Test

```
$ npm test

> kafka@1.0.0 test /home/vagrant/kafka
> ava -v -u


  ✔ producer test (291ms)
  ✔ consumer test

  2 tests passed
```

#### Kafka Producer

-  src/producer/app.js 

```
"use strict";
var kafka = require("kafka-node");

const Producer = kafka.HighLevelProducer;
const client = new kafka.KafkaClient({
    kafkaHost: "192.168.33.50:9092"
});
const producer = new Producer(client, {
    partitionerType: 1
});

producer.on("ready", () => {
    const payloads = [
        {
            topic: "mytopic-1",
            messages: JSON.stringify({name: "tanaka takeshi", age: 24, sex: "M"})
        }
        ,{
            topic: "mytopic-2",
            messages: [
                JSON.stringify({name: "suzuki aiko", age: 20, sex: "F"}),
                JSON.stringify({name: "yamashita yuji", age: 28, sex: "M"})
            ]
        }
    ];

    let sends = 0;

    producer.send(payloads, (err, data) => {
        if (err) console.log(err);
        else console.log('send %d messages', ++sends);
        process.exit();
    });
});

```

#### Kafka Consumer

-  src/consumer/app.js 

```
'use strict';
var kafka = require('kafka-node');

const Consumer = kafka.Consumer;
const client = new kafka.KafkaClient({kafkaHost: "192.168.33.50:9092"});
const consumer = new Consumer(
    client,
    [{topic: "mytopic-2", partision:0}],
    {
        groupId: "simple-consumer",
        autoCommit: true,
        fromOffset: true
    }
);

consumer.on("message", (message, err) => {
    if (err) console.log("error : " + err);

    const json = JSON.parse(message.value);
    console.log("JSON:" + JSON.stringify(json));
    console.log("Name:" + json.name);
    console.log("Age:" + json.age);
    console.log("Sex Type:" + json.sex);
});

consumer.on('error', function (err) {
    console.log('error', err);
});

```

#### 終わりに
本格的に使う機会がないので取りあえずNodejsで動かしてみただけっす。


#### 参考
[apache kafka quickstart(kafkaの基本的な使い方）](https://kafka.apache.org/quickstart)
[kafka-node](https://www.npmjs.com/package/kafka-node#producer)