B.80. KafkaConnectSpec スキーマ参照

以下で使用されます。KafkaConnect

Kafka Connect クラスターを設定します。

B.80.1. 設定

Kafkaのオプションをキーとして設定するには、configプロパティを使用します。

標準の Apache Kafka Connect 設定が提供されることがありますが、AMQ Streams によって直接管理されないプロパティーに限定されます。

以下に関連する設定オプションは設定できません。

  • Kafka クラスターブートストラップアドレス
  • セキュリティー (暗号化、認証、および承認)
  • リスナー / REST インターフェースの設定
  • プラグインパスの設定

値は以下の 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 プロパティーになかった場合に自動的に設定されます。

禁止されているオプションには例外があります。TLS バージョンの特定の 暗号スイート を使用して、クライアント接続に許可される 3 つの ssl 設定オプションを使用できます。暗号スイートは、セキュアな接続とデータ転送のためのアルゴリズムを組み合わせます。ssl.endpoint.identification.algorithm プロパティーを設定して、ホスト名の検証を有効または無効にすることもできます。

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"
    ssl.enabled.protocols: "TLSv1.2"
    ssl.protocol: "TLSv1.2"
    ssl.endpoint.identification.algorithm: HTTPS
  # ...

TLSバージョンの特定の暗号スイートを使用するクライアント接続のために、許可されたsslプロパティを設定することができます。また、 ssl.endpoint.identification.algorithmプロパティを設定して、ホスト名の検証を有効または無効にすることもできます。