Kafkaのメッセージにキーを付ける方法（ksqlDB編）

Kafkaのメッセージはヘッダー、キー、バリューで構成されています。
キーは、メッセージを一意に識別したい場合、Topic内でキー単位で最新のメッセージを有効にしたい場合、Topicのメッセージを元にDB内のレコードをUPDATE・DELETEしたい場合などに必要になるので重要です。

ソースとなるサービスやコネクターがメッセージのキーを扱わないものの場合、何もしなければKafkaにキーのないメッセージが送られてくることになります。
以下では、ksqlDBのストリームを使用してメッセージにキーを付与する方法を示しています。
データベースとしてDb2を、kafkaはConfluentを使用した例です。

## やろうとしていること
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/b8b509c9-0362-3105-cd55-e96f01b75e69.png)

## 1. Topic1にあるキーのないメッセージに対してStream1を作成する

Topic1の内容は以下の通りです。
キーがないことが確認できます。
```sh:Topic1内のメッセージ
ksql> print 'topic1' from beginning;
Key format: ¯\_(ツ)_/¯ - no data processed
Value format: AVRO or KAFKA_STRING
rowtime: 2023/08/14 06:38:11.074 Z, key: <null>, value: {"col1": "a1", "col2": "a", "col3": "test1"}, partition: 0
rowtime: 2023/08/14 06:38:16.039 Z, key: <null>, value: {"col1": "a2", "col2": "aa", "col3": "test2"}, partition: 0
rowtime: 2023/08/14 06:38:27.083 Z, key: <null>, value: {"col1": "a3", "col2": "aaa", "col3": "test3"}, partition: 0
```
ksqlDBのストリームでメッセージを加工するためには加工元のTopicに対するストリームを作成し、加工処理を行うストリームから参照できるようにする必要があります。
加工元のTopicに対するStream1の定義内容は以下の通りです。
```sql:Stream1の定義
CREATE STREAM STREAM1
  WITH (KAFKA_TOPIC='topic1', VALUE_FORMAT='AVRO');
```
```sql:Stream1内のメッセージ
ksql> SELECT * FROM STREAM1;
+-------------------------------+-------------------------------+-------------------------------+
|COL1                           |COL2                           |COL3                           |
+-------------------------------+-------------------------------+-------------------------------+
|a1                             |a                              |test1                          |
|a2                             |aa                             |test2                          |
|a3                             |aaa                            |test3                          |
```

## 2. Stream1のメッセージを取得してメッセージにキーを追加するStream2を作成する

加工処理を行うストリームを作成します。
ストリームの定義の中で加工したメッセージを書き込むTopicを指定します。
```sql:Stream2の定義
CREATE STREAM STREAM2
  WITH (KAFKA_TOPIC='topic2', KEY_FORMAT='AVRO', VALUE_FORMAT='AVRO')
  AS SELECT
    STRUCT(COL1:=STREAM1.COL1) AS COL1,
    STREAM1.COL2,
    STREAM1.COL3
  FROM
    STREAM1
  PARTITION BY
    STRUCT(COL1:=STREAM1.COL1)
  EMIT CHANGES;
```
```sql:Stream2内のメッセージ
ksql> SELECT * FROM STREAM2;
+-------------------------------+-------------------------------+-------------------------------+
|COL1                           |COL2                           |COL3                           |
+-------------------------------+-------------------------------+-------------------------------+
|{COL1=a1}                      |a                              |test1                          |
|{COL1=a2}                      |aa                             |test2                          |
|{COL1=a3}                      |aaa                            |test3                          |
```

Stream2によってTopic2にキーがあるメッセージが書き込まれていることが確認できます。
```sh:Topic2内のメッセージ
ksql> print 'topic2' from beginning;
Key format: AVRO or HOPPING(KAFKA_STRING) or TUMBLING(KAFKA_STRING) or KAFKA_STRING
Value format: AVRO or KAFKA_STRING
rowtime: 2023/08/14 06:38:11.074 Z, key: {"COL1": "a1"}, value: {"COL2": "a", "COL3": "test1"}, partition: 0
rowtime: 2023/08/14 06:38:16.039 Z, key: {"COL1": "a2"}, value: {"COL2": "aa", "COL3": "test2"}, partition: 0
rowtime: 2023/08/14 06:38:27.083 Z, key: {"COL1": "a3"}, value: {"COL2": "aaa", "COL3": "test3"}, partition: 0
```
Topic2のスキーマからもキーがあることを確認することができます。
```sh:Topic2のスキーマサブジェクト（キー）
$ curl -sX GET http://localhost:8081/subjects/topic2-key/versions/latest | jq .schema -r | jq
{
  "type": "record",
  "name": "Stream2Key",
  "namespace": "io.confluent.ksql.avro_schemas",
  "fields": [
    {
      "name": "COL1",
      "type": [
        "null",
        "string"
      ],
      "default": null
    }
  ]
}
```

```sh:Topic2のスキーマサブジェクト（バリュー）
$ curl -sX GET http://localhost:8081/subjects/topic2-value/versions/latest | jq .schema -r | jq
{
  "type": "record",
  "name": "KsqlDataSourceSchema",
  "namespace": "io.confluent.ksql.avro_schemas",
  "fields": [
    {
      "name": "COL2",
      "type": [
        "null",
        "string"
      ],
      "default": null
    },
    {
      "name": "COL3",
      "type": [
        "null",
        "string"
      ],
      "default": null
    }
  ]
}
```
キーのないメッセージをksqlDBのストリーム処理によって加工し、キーのあるメッセージをTopicに書き込むことができました。キーのあるメッセージが書き込まれたTopicであれば、JDBC Sink Connectorを使用することで後続の連携先DB内のテーブルに対してメッセージをもとにINSERTだけでなくUPDATEやDELETEができるようになります。

ksqlDBであれば、プログラミング言語でアプリケーションを実装することなく、SQLライクな実装で簡単にこのようなことが実現できます。