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

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

キーがない場合の例1

JDBC Source Connectorは、主キーがあるテーブルから取得したデータをキーがないメッセージとしてTopicに書き込みます。

ソースとなるテーブルの構成（IDが主キー）

$ db2 describe table sample_txt

                                データ・タイ データ・         列の       スケ
列名                            プ・スキーマ タイプ名         長さ       ール  NULL
------------------------------- --------- ------------------- ---------- ----- ------
ID                              SYSIBM    INTEGER                      4     0 いいえ
TEXT1                           SYSIBM    VARCHAR                      5     0 はい  
TEXT2                           SYSIBM    VARCHAR                     10     0 はい  
TEXT3                           SYSIBM    VARCHAR                     15     0 はい

ソースとなるテーブルのデータ（IDが主キー）

$ db2 "select * from sample_txt"

ID          TEXT1 TEXT2      TEXT3          
----------- ----- ---------- ---------------
          1 a     b          c              
          2 aa    bb         cc             
          3 aaa   bbb        ccc

JDBC Source Connectorの作成コマンド（IDが主キー）

curl -sX PUT http://localhost:8083/connectors/sample_txt_connector/config \
-H "Content-Type: application/json" \
--data '{
"name":"sample_txt_connector",
"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max":"1",
"key.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url":"http://localhost:8081",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url":"http://localhost:8081",
"connection.url":"jdbc:db2://xxxxx:50000/dbname",
"connection.user":"******",
"connection.password":"*********",
"mode":"bulk",
"query":"select * from schema1.sample_txt",
"poll.interval.ms":"1000000",
"topic.prefix":"sample_txt_topic"
}' | jq

Topicに書き込まれたメッセージ（IDが主キー、だがメッセージにキーがない）

ksql> print 'sample_txt_topic' from beginning;
Key format: ¯\_(ツ)_/¯ - no data processed
Value format: AVRO or KAFKA_STRING
rowtime: 2023/08/13 20:50:40.163 Z, key: <null>, value: {"ID": 1, "TEXT1": "a", "TEXT2": "b", "TEXT3": "c"}, partition: 0
rowtime: 2023/08/13 20:50:40.164 Z, key: <null>, value: {"ID": 2, "TEXT1": "aa", "TEXT2": "bb", "TEXT3": "cc"}, partition: 0
rowtime: 2023/08/13 20:50:40.164 Z, key: <null>, value: {"ID": 3, "TEXT1": "aaa", "TEXT2": "bbb", "TEXT3": "ccc"}, partition: 0

JDBC Source Connectorの対象とするテーブルに主キーがあっても、取得したデータをもとにTopicに書き込むメッセージにはキーがないことがわかります。

キーがない場合の例2

JDBC Source Connectorは、主キーがないテーブルから取得したデータをキーがないメッセージとしてTopicに書き込みます。

ソースとなるテーブルの構成（主キーなし）

$ db2 describe table sample_txt_nokey

                                データ・タイ データ・         列の       スケ
列名                            プ・スキーマ タイプ名         長さ       ール  NULL
------------------------------- --------- ------------------- ---------- ----- ------
TEXT1                           SYSIBM    VARCHAR                      5     0 はい  
TEXT2                           SYSIBM    VARCHAR                     10     0 はい  
TEXT3                           SYSIBM    VARCHAR                     15     0 はい

ソースとなるテーブルのデータ（主キーなし）

db2 "select * from sample_txt_nokey"

TEXT1 TEXT2      TEXT3          
----- ---------- ---------------
x     y          z              
xx    yy         zz             
xxx   yyy        zzz

JDBC Source Connectorの作成コマンド（主キーなし）

curl -sX PUT http://localhost:8083/connectors/sample_txt_nokey_connector/config \
-H "Content-Type: application/json" \
--data '{
"name":"sample_txt_nokey_connector",
"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max":"1",
"key.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url":"http://localhost:8081",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url":"http://localhost:8081",
"connection.url":"jdbc:db2://xxxxx:50000/dbname",
"connection.user":"******",
"connection.password":"*********",
"mode":"bulk",
"query":"select * from schema1.sample_txt_nokey",
"poll.interval.ms":"1000000",
"topic.prefix":"sample_txt_nokey_topic"
}' | jq

メッセージにキーがない場合、key.converterやkey.converter.schema.registry.urlを指定してもエラーはなりませんが使用されることもありません。

Topicに書き込まれたメッセー（キーなし）

ksql> print 'sample_txt_nokey_topic' from beginning;
Key format: ¯\_(ツ)_/¯ - no data processed
Value format: AVRO or KAFKA_STRING
rowtime: 2023/08/13 21:01:33.682 Z, key: <null>, value: {"TEXT1": "x", "TEXT2": "y", "TEXT3": "z"}, partition: 0
rowtime: 2023/08/13 21:01:33.683 Z, key: <null>, value: {"TEXT1": "xx", "TEXT2": "yy", "TEXT3": "zz"}, partition: 0
rowtime: 2023/08/13 21:01:33.683 Z, key: <null>, value: {"TEXT1": "xxx", "TEXT2": "yyy", "TEXT3": "zzz"}, partition: 0

メッセージにキーがないことがわかります。

メッセージにキーを付与する例

SMT（Single Message Transforms）を使えば簡単にコネクターを通過するメッセージを加工することができます。ここではメッセージにキーを付与するValueToKeyを使用します。

JDBC Source Connectorの作成コマンド（SMTでキーを追加）

curl -sX PUT http://localhost:8083/connectors/sample_txt_addkey_connector/config \
-H "Content-Type: application/json" \
--data '{
"name":"sample_txt_addkey_connector",
"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max":"1",
"key.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url":"http://localhost:8081",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url":"http://localhost:8081",
"connection.url":"jdbc:db2://vm1:50000/dbname",
"connection.user":"******",
"connection.password":"*********",
"mode":"bulk",
"query":"select * from schema1.sample_txt_nokey",
"poll.interval.ms":"1000000",
"topic.prefix":"sample_txt_addkey_topic",
"transforms":"ValueToKey",
"transforms.ValueToKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
"transforms.ValueToKey.fields":"TEXT1"
}' | jq

transformsで始まる3行がSMTに該当するプロパティです。これを追加するだけでメッセージにキーが追加できます。
この例ではTEXT1という項目をキーに設定しています。

Topicに書き込まれたメッセージ（SMTでキーを追加）

ksql> print 'sample_txt_addkey_topic' from beginning;
Key format: AVRO or KAFKA_STRING
Value format: AVRO or KAFKA_STRING
rowtime: 2023/08/13 21:19:52.400 Z, key: {"TEXT1": "x"}, value: {"TEXT1": "x", "TEXT2": "y", "TEXT3": "z"}, partition: 0
rowtime: 2023/08/13 21:19:52.403 Z, key: {"TEXT1": "xx"}, value: {"TEXT1": "xx", "TEXT2": "yy", "TEXT3": "zz"}, partition: 0
rowtime: 2023/08/13 21:19:52.403 Z, key: {"TEXT1": "xxx"}, value: {"TEXT1": "xxx", "TEXT2": "yyy", "TEXT3": "zzz"}, partition: 0

これでメッセージにキーが追加されました。