3.3.5. Kafka Connect の設定
AMQ Streams では、Apache Kafka ドキュメント に記載されている特定のオプションを編集して、Apache Kafka Connect ノードの設定をカスタマイズできます。
以下に関連する設定オプションは設定できません。
- Kafka クラスターブートストラップアドレス
- セキュリティー (暗号化、認証、および承認)
- リスナー / REST インターフェイスの設定
- プラグインパスの設定
これらのオプションは AMQ Streams によって自動的に設定されます。
3.3.5.1. Kafka Connect の設定
Kafka Connect は、KafkaConnect.spec および KafkaConnectS2I.spec の config プロパティーを使用して設定されます。このプロパティーには、Kafka Connect 設定オプションがキーとして含まれます。値は以下の JSON タイプのいずれかになります。
- 文字列
- 数値
- ブール値
AMQ Streams で直接管理されるオプションを除き、Apache Kafka ドキュメント に記載されているオプションを指定および設定できます。以下の文字列の 1 つと同じキーまたは以下の文字列の 1 つで始まるキーを持つ設定オプションは禁止されています。
-
ssl. -
sasl. -
security. -
listeners -
plugin.path -
rest. -
bootstrap.servers
禁止されているオプションが config プロパティーにある場合、そのオプションは無視され、警告メッセージが Cluster Operator ログファイルに出力されます。その他のオプションはすべて Kafka Connect に渡されます。
提供された config オブジェクトのキーまたは値は Cluster Operator によって検証されません。無効な設定を指定すると、Kafka Connect クラスターが起動しなかったり、不安定になる可能性があります。この状況で、KafkaConnect.spec.config または KafkaConnectS2I.spec.config オブジェクトの設定を修正すると、Cluster Operator は新しい設定をすべての Kafka Connect ノードにロールアウトできます。
以下のオプションにはデフォルト値があります。
-
group.id、デフォルト値connect-cluster -
offset.storage.topic、デフォルト値connect-cluster-offsets -
config.storage.topic、デフォルト値connect-cluster-configs -
status.storage.topic、デフォルト値connect-cluster-status -
key.converter、デフォルト値org.apache.kafka.connect.json.JsonConverter -
value.converter、デフォルト値org.apache.kafka.connect.json.JsonConverter
これらのオプションは、KafkaConnect.spec.config または KafkaConnectS2I.spec.config プロパティーになかった場合に自動的に設定されます。
許可される 3 つの ssl 設定オプションを使用して、TLS バージョンの固有の暗号スイートで外部リスナーを実行します。暗号スイートでは、セキュアな接続とデータ転送のためのアルゴリズムが組み合わされます。
Kafka Connect の設定例
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
name: my-connect
spec:
# ...
config:
group.id: my-connect-cluster
offset.storage.topic: my-connect-cluster-offsets
config.storage.topic: my-connect-cluster-configs
status.storage.topic: my-connect-cluster-status
key.converter: org.apache.kafka.connect.json.JsonConverter
value.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: true
value.converter.schemas.enable: true
config.storage.replication.factor: 3
offset.storage.replication.factor: 3
status.storage.replication.factor: 3
ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 1
ssl.enabled.protocols: "TLSv1.2" 2
ssl.protocol: "TLSv1.2" 3
# ...
