1. Home
  2. Platform
  3. Kafka 用アプリケーションの構築
  4. MQTT Proxy

MQTT Proxy の構成オプション

ネットワーク

リスナー

リスナーのリスト。各リスナーには、ホスト名およびポートが含まれている必要があります。例: tcp://myhost:1883、https://0.0.0.0:8081

  • 型: string
  • デフォルト : 0.0.0.0:1883
  • 指定可能な値 : host1:port1, host2:port2, ... という形式でのホストとポートのペアのリスト
  • 重要度: 高
network.threads.num

ネットワーク IO を処理するために Confluent MQTT Proxy によりデプロイされるスレッド数。

  • 型: int
  • デフォルト: 1
  • 指定可能な値: [1,…]
  • 重要度: 高
network.epoll.enabled

Linux 上での受信接続に対して、標準の NIO ではなく epoll を使用します。epoll が使用可能ではない場合に、このプロパティを有効にすると、構成エラーが発生します。

  • 型: ブール値
  • デフォルト: false
  • 指定可能な値: [false、true]
  • 重要度: 高

プロトコル

mqtt.message.max.bytes

バイト単位での最大許容 MQTT メッセージサイズ（固定ヘッダーを除く）。

  • 型: int
  • デフォルト: 8092
  • 指定可能な値: [0,…]
  • 重要度: 中
listeners.security.protocol

MQTT クライアントと Confluent MQTT Proxy 間で使用されるプロトコル。

  • 型: string
  • デフォルト: PLAINTEXT
  • 指定可能な値: [PLAINTEXT, SSL, TLS, SASL_PLAINTEXT, SASL_SSL, SASL_TLS]
  • 重要度: 高

セキュリティ、認証、および暗号化の通信設定の詳細については、「通信セキュリティ設定」を参照してください。

listeners.ssl.protocol

SSLContext の生成に使用する TLS/SSL プロトコル。デフォルト設定は TLS であり、ほとんどのケースに適しています。最新の Java 仮想マシンで使用できる値は、TLS、TLSv1.1、TLSv1.2 です。古い JVM では SSL、SSLv2、および SSLv3 がサポートされている場合もありますが、セキュリティに関する既知の脆弱性があるため、使用しないことをお勧めします。

  • 型: string
  • デフォルト: TLS
  • 重要度: 中
listeners.ssl.provider

TLS/SSL 接続に使用するセキュリティプロバイダーの名前。デフォルト値は、Java 仮想マシンのデフォルトのセキュリティプロバイダーです。

  • 型: string
  • デフォルト: null
  • 重要度: 中
listeners.ssl.cipher.suites

暗号スイートのリストです。これは、認証、暗号化、MAC、およびキー交換アルゴリズムの名前付き組み合わせで、TLS または SSL ネットワークプロトコルを使用したネットワーク接続のセキュリティ設定をネゴシエートするために使用されます。デフォルトでは、使用可能なすべての暗号スイートがサポートされます。

  • 型: list
  • デフォルト: null
  • 重要度: 低
listeners.ssl.enabled.protocols

TLS/SSL 接続で有効なプロトコルのリストです。

  • 型: list
  • デフォルト: TLSv1.2、TLSv1.1、TLSv1
  • 重要度: 中
listeners.ssl.keystore.type

キーストアファイルのファイルフォーマット。クライアントでは省略可能です。

  • 型: string
  • デフォルト: JKS
  • 重要度: 中
listeners.ssl.keystore.location

キーストアファイルの場所。クライアントでは省略可能です。クライアントの相互認証に使用できます。

  • 型: string
  • デフォルト: null
  • 重要度: 高
listeners.ssl.keystore.password

キーストアファイルのストアパスワード。クライアントでは省略可能です。ssl.keystore.location を構成した場合にのみ必要となります。

  • 型: password
  • デフォルト: null
  • 重要度: 高
listeners.ssl.key.password

キーストアファイル内のプライベートキーのパスワード。クライアントでは省略可能です。

  • 型: password
  • デフォルト: null
  • 重要度: 高
listeners.ssl.truststore.type

トラストストアファイルのファイルフォーマット。

  • 型: string
  • デフォルト: JKS
  • 重要度: 中
listeners.ssl.truststore.location

トラストストアファイルの場所。

  • 型: string
  • デフォルト: null
  • 重要度: 高
listeners.ssl.truststore.password

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアにアクセスできますが、整合性チェックが無効になります。

  • 型: password
  • デフォルト: null
  • 重要度: 高
listeners.ssl.keymanager.algorithm

TLS/SSL 接続のキーマネージャーファクトリで使用されるアルゴリズム。デフォルト値は、Java 仮想マシン用に構成されているキーマネージャーファクトリアルゴリズムです。

  • 型: string
  • デフォルト: SunX509
  • 重要度: 低
listeners.ssl.trustmanager.algorithm

TLS/SSL 接続のトラストマネージャーファクトリで使用されるアルゴリズム。デフォルト値は、Java 仮想マシン用に構成されているトラストマネージャーファクトリアルゴリズムです。

  • 型: string
  • デフォルト: PKIX
  • 重要度: 低
listeners.ssl.endpoint.identification.algorithm

サーバー証明書を使用してサーバーホスト名を検証するエンドポイント識別アルゴリズム。

  • 型: string
  • デフォルト: https
  • 重要度: 低
listeners.ssl.secure.random.implementation

TLS/SSL 暗号化操作に使用する SecureRandom PRNG 実装。

  • 型: string
  • デフォルト: null
  • 重要度: 低
listeners.ssl.client.auth

クライアント認証を要求するようサーバーを構成します。

  • 型: string
  • デフォルト: none
  • 指定可能な値: [none, requested, required]
  • 重要度: 中

ストリーム

bootstrap.servers

Apache Kafka® クラスターへの初期接続の確立に使用するための、ホストとポートのペアのリスト。ここでブートストラップ用にどのサーバーが指定されているかにかかわらず、クライアントはすべてのサーバーを使用します。このリストは、すべてのサーバーを検出するために使用する最初のホストにのみ影響します。このリストは、host1:port1,host2:port2,... という形式にする必要があります。これらのサーバーは、初期接続ですべてのクラスターメンバーシップを検出するためにのみ使用されます。これは動的に変わる可能性があるので、このリストにすべてのサーバーセットを含める必要はありません（ただし、サーバーの障害に備えて、複数指定しておくこともできます）。

  • 型: string
  • 重要度: 高
topic.regex.list

MQTT トピックを Kafka トピックにマッピングするために使用される <kafka topic> :<regex> 型のペアのコンマ区切りのリスト。

  • 型: list
  • 指定可能な値 : <kafka topic1> :<regex1>, <kafka topic2>:<regex2>, ... という形式でのペアのリスト
  • 重要度: 高
stream.threads.num

レコードを Kafka にパブリッシュするスレッド数

  • 型: int
  • デフォルト: 1
  • 指定可能な値: [1,…]
  • 重要度: 高
producer.buffer.memory

サーバーへの送信待機中のレコードをバッファするためにプロデューサーが使用できるメモリーの合計バイト数。サーバーに配信できるよりも早い時点でレコードが送信された場合、プロデューサーは、その時点から max.block.ms の間にわたりブロックし、例外をスローします。この設定は、プロデューサーが使用できる合計メモリーとほぼ一致しますが、プロデューサーが使用する全メモリーがバッファリングのために使用されるわけではないため、厳格な制限ではありません。一部のメモリーはさらに、送信リクエストだけではなく圧縮のために使用されます（圧縮が有効な場合）。

  • 型: long
  • デフォルト: 33554432
  • 指定可能な値: [0,…]
  • 重要度: 高
producer.compression.type

プロデューサーにより生成されるすべてのデータの圧縮タイプ。デフォルトは none（つまり、圧縮なし）です。指定可能な値は none、gzip、snappy、または lz4 です。圧縮はデータのバッチ全体に関するものであるため、バッチ処理の有効性も圧縮率に影響します（つまり、バッチ処理が多くなるほど、圧縮が向上します）。

  • 型: string
  • デフォルト: none
  • 重要度: 高
producer.batch.size

複数のレコードが同じパーティションに送信されるたびに、プロデューサーはレコードをまとめてバッチ処理することにより、リクエストを少なくしようと試みます。これは、クライアントとサーバーの両方でのパフォーマンス向上に役立ちます。この構成により、バイト単位でのデフォルトのバッチサイズが制御されます。このサイズより大きなレコードのバッチ処理は試行されません。ブローカーに送信されるリクエストには複数のバッチが含まれるようになり、送信のために使用可能なデータが各パーティションの 1 つのバッチに含まれます。小さなバッチサイズではバッチ処理の頻度が少なくなり、スループットが低下する場合があります（ゼロのバッチサイズでは、バッチ処理が完全に無効になります）。追加のレコードを想定したうえで、指定したバッチサイズのバッファを常に割り当てるため、非常に大きなバッチサイズでは、メモリーが多少無駄に使用される場合があります。

  • 型: int
  • デフォルト: 16384
  • 指定可能な値: [0,…]
  • 重要度: 中
producer.linger.ms

プロデューサーは、リクエストの送信から次の送信までに到着したすべてのレコードをまとめて、バッチ処理された単一のリクエストにグループ化します。通常、これは、負荷がかかった状況、つまり、レコードが送信よりも早く到着した場合にのみ行われます。ただし、状況によっては、中程度の負荷が生じる場合であっても、クライアントではリクエストの数を減らすことが望ましい場合があります。この設定では、人為的な短い遅延を追加することにより、これを実現できます。つまり、レコードをすぐに送信するのではなく、プロデューサーは他のレコードを送信できるように所定の遅延まで待機し、送信をまとめてバッチ処理できるようにします。これは、TCP における Nagle アルゴリズムと同様であると考えることができます。この設定により、バッチ処理のための遅延の上限が指定され、パーティションにおいて batch.size に相当するレコードが到着すると、この設定に関係なくレコードが送信されます。ただし、そのパーティションにおいて累積したバイト数が、この上限よりも少ない場合は、追加のレコードが到着するまで待ちながら、指定した時間にわたり「待機」します。この設定はデフォルトでは 0（つまり、遅延なし）になります。たとえば、linger.ms=5 と設定した場合、送信されるリクエスト数を減少させる効果があると予想されますが、負荷がない状態で送信されるレコードにおいて、最大 5 ms のレイテンシが追加で生じると予想されます。

  • 型: long
  • デフォルト: 0
  • 指定可能な値: [0,…]
  • 重要度: 中
producer.client.id

リクエストを行う際にサーバーに渡す ID 文字列。これは、論理アプリケーション名をサーバー側のリクエストログに含めることで、IP とポートだけでなく、リクエストのソースを追跡できるようにすることが目的です。

  • 型: string
  • デフォルト: ""
  • 重要度: 中
producer.send.buffer.bytes

データを送信する際に使用される TCP 送信バッファ（SO_SNDBUF）のサイズ。値が -1 である場合は、OS のデフォルトが使用されます。

  • 型: int
  • デフォルト: 131072
  • 指定可能な値: [-1、...]
  • 重要度: 中
producer.receive.buffer.bytes

データを読み取る際に使用される TCP 受信バッファ（SO_RCVBUF）のサイズ。値が -1 である場合は、OS のデフォルトが使用されます。

  • 型: int
  • デフォルト: 32768
  • 指定可能な値: [-1、...]
  • 重要度: 中
producer.max.request.size

バイト単位でのリクエストの最大サイズ。膨大なリクエストの送信を回避するため、この設定により、単一のリクエストにおいてプロデューサーが送信するレコードバッチ数が制限されます。これは実質的には、最大レコードバッチサイズの上限でもあります。サーバーにはレコードバッチサイズの独自の上限があり、これとは異なる場合があることに注意してください。

  • 型: int
  • デフォルト: 1048576
  • 指定可能な値: [0,…]
  • 重要度: 中
producer.reconnect.backoff.ms

特定のホストへの再接続を試行するまでの基本待機時間。これにより、ホストへの接続が短時間に何度も繰り返されることを回避できます。このバックオフは、クライアントによるブローカーへの接続のすべての試行に対して適用されます。

  • 型: long
  • デフォルト: 50
  • 指定可能な値: [0,…]
  • 重要度: 低
producer.reconnect.backoff.max.ms

繰り返し接続に失敗しているブローカーに再接続する場合に待機するための、ミリ秒単位での最大時間。これを指定した場合は、連続して接続に失敗するたびに、ホストあたりのバックオフが、この最大値まで指数関数的に増加します。接続ストームを回避するため、バックオフ増加の計算後には 20% のランダムジッターが追加されます。

  • 型: long
  • デフォルト: 1000
  • 指定可能な値: [0,…]
  • 重要度: 低
producer.max.block.ms

この構成により、KafkaProducer.send() および KafkaProducer.partitionsFor() がブロックする時間の長さが制御されます。バッファがいっぱいであること、またはメタデータが使用不可であることを理由として、これらのメソッドをブロックすることができます。ユーザーが提供したシリアライザーまたはパーティショナーでのブロックは、このタイムアウトに対してカウントされません。

  • 型: long
  • デフォルト: 60000
  • 指定可能な値: [0,…]
  • 重要度: 中
producer.request.timeout.ms

この構成では、クライアントがリクエストの応答を待つ最大の待機時間を指定します。タイムアウト時間が経過するまで応答を受信できなかった場合、クライアントは、必要に応じてリクエストを再送し、再試行回数が上限に達した場合は、リクエストを失敗とします。プロデューサーの不要な再試行によるメッセージ重複の可能性を低減するため、これは replica.lag.time.max.ms（ブローカー構成）よりも大きくする必要があります。

  • 型: int
  • デフォルト: 30000
  • 指定可能な値: [0,…]
  • 重要度: 中
producer.metadata.max.age.ms

ここで指定された時間（単位: ミリ秒）が経過すると、メタデータが強制的に更新されます。これは、パーティションのリーダーシップに変更がなくても、新しいブローカーやパーティションがあれば積極的に検出できるようにするためです。

  • 型: long
  • デフォルト: 300000
  • 指定可能な値: [0,…]
  • 重要度: 低
producer.metrics.sample.window.ms

メトリクスサンプルが計算される時間枠。

  • 型: long
  • デフォルト: 30000
  • 指定可能な値: [0,…]
  • 重要度: 低
producer.metrics.num.samples

メトリクスの計算用に維持されるサンプルの数。

  • 型: int
  • デフォルト: 2
  • 指定可能な値: [1,…]
  • 重要度: 低
producer.metrics.recording.level

メトリクスの最高記録レベル。

  • 型: string
  • デフォルト: INFO
  • 指定可能な値: [INFO、DEBUG]
  • 重要度: 低
producer.metric.reporters

Metrics Reporter として使用するクラスのリスト。org.apache.kafka.common.metrics.MetricsReporter インターフェイスを実装することで、新しいメトリック作成の通知を受けるクラスをプラグインすることができます。JMX 統計情報を登録するための JmxReporter が必ず含まれます。

producer.max.in.flight.requests.per.connection

ブロックまでに単一の接続においてクライアントが送信する、確認応答されていないリクエストの最大数。この設定が 1 より大きく設定されている場合に、送信の失敗が発生すると、再試行によるメッセージ再順序付けのリスクが生じることに注意してください（再試行が有効な場合）。

  • 型: int
  • デフォルト: 5
  • 指定可能な値: [1,…]
  • 重要度: 低
producer.connections.max.idle.ms

アイドル状態の接続は、この構成で指定された時間（ミリ秒）の経過後に終了されます。

  • 型: long
  • デフォルト: 540000
  • 重要度: 中
producer.partitioner.class

org.apache.kafka.clients.producer.Partitioner インターフェイスを実装するパーティショナークラス。

  • 型: class
  • デフォルト: org.apache.kafka.clients.producer.internals.DefaultPartitioner
  • 重要度: 中
producer.interceptor.classes

インターセプターとして使用するクラスのリスト。org.apache.kafka.clients.producer.ProducerInterceptor インターフェイスを実装すると、Kafka クラスターにパブリッシュされる前に、プロデューサーが受信したレコードをインターセプト（および場合によっては変換）することができます。デフォルトでは、インターセプターはありません。

producer.security.protocol

ブローカーとの通信に使用されるプロトコル。指定可能な値は、PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL です。

  • 型: string
  • デフォルト: PLAINTEXT
  • 重要度: 中
producer.ssl.protocol

SSLContext の生成に使用する TLS/SSL プロトコル。デフォルト設定は TLS であり、ほとんどのケースに適しています。最新の Java 仮想マシンで使用できる値は、TLS、TLSv1.1、TLSv1.2 です。古い JVM では SSL、SSLv2、および SSLv3 がサポートされている場合もありますが、セキュリティに関する既知の脆弱性があるため、使用しないことをお勧めします。

  • 型: string
  • デフォルト: TLS
  • 重要度: 中
producer.ssl.provider

TLS/SSL 接続に使用するセキュリティプロバイダーの名前。デフォルト値は、Java 仮想マシンのデフォルトのセキュリティプロバイダーです。

  • 型: string
  • デフォルト: null
  • 重要度: 中
producer.ssl.cipher.suites

暗号スイートのリストです。これは、認証、暗号化、MAC、およびキー交換アルゴリズムの名前付き組み合わせで、TLS または SSL ネットワークプロトコルを使用したネットワーク接続のセキュリティ設定をネゴシエートするために使用されます。デフォルトでは、使用可能なすべての暗号スイートがサポートされます。

  • 型: list
  • デフォルト: null
  • 重要度: 低
producer.ssl.enabled.protocols

TLS/SSL 接続で有効なプロトコルのリストです。

  • 型: list
  • デフォルト: TLSv1.2、TLSv1.1、TLSv1
  • 重要度: 中
producer.ssl.keystore.type

キーストアファイルのファイルフォーマット。クライアントでは省略可能です。

  • 型: string
  • デフォルト: JKS
  • 重要度: 中
producer.ssl.keystore.location

キーストアファイルの場所。クライアントでは省略可能です。クライアントの相互認証に使用できます。

  • 型: string
  • デフォルト: null
  • 重要度: 高
producer.ssl.keystore.password

キーストアファイルのストアパスワード。クライアントでは省略可能です。ssl.keystore.location を構成した場合にのみ必要となります。

  • 型: password
  • デフォルト: null
  • 重要度: 高
producer.ssl.key.password

キーストアファイル内のプライベートキーのパスワード。クライアントでは省略可能です。

  • 型: password
  • デフォルト: null
  • 重要度: 高
producer.ssl.truststore.type

トラストストアファイルのファイルフォーマット。

  • 型: string
  • デフォルト: JKS
  • 重要度: 中
producer.ssl.truststore.location

トラストストアファイルの場所。

  • 型: string
  • デフォルト: null
  • 重要度: 高
producer.ssl.truststore.password

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアにアクセスできますが、整合性チェックが無効になります。

  • 型: password
  • デフォルト: null
  • 重要度: 高
producer.ssl.keymanager.algorithm

TLS/SSL 接続のキーマネージャーファクトリで使用されるアルゴリズム。デフォルト値は、Java 仮想マシン用に構成されているキーマネージャーファクトリアルゴリズムです。

  • 型: string
  • デフォルト: SunX509
  • 重要度: 低
producer.ssl.trustmanager.algorithm

TLS/SSL 接続のトラストマネージャーファクトリで使用されるアルゴリズム。デフォルト値は、Java 仮想マシン用に構成されているトラストマネージャーファクトリアルゴリズムです。

  • 型: string
  • デフォルト: PKIX
  • 重要度: 低
producer.ssl.endpoint.identification.algorithm

サーバー証明書を使用してサーバーホスト名を検証するエンドポイント識別アルゴリズム。

  • 型: string
  • デフォルト: https
  • 重要度: 低
producer.ssl.secure.random.implementation

SSL 暗号化操作に使用する SecureRandom PRNG 実装。

  • 型: string
  • デフォルト: null
  • 重要度: 低
producer.sasl.kerberos.service.name

Kafka が実行される際の Kerberos プリンシパル名。これは、Kafka の JAAS 構成または Kafka の構成のどちらかで定義できます。

  • 型: string
  • デフォルト: null
  • 重要度: 中
producer.sasl.kerberos.kinit.cmd

Kerberos の kinit コマンドパス。

  • 型: string
  • デフォルト: /usr/bin/kinit
  • 重要度: 低
producer.sasl.kerberos.ticket.renew.window.factor

最後の更新からチケットの有効期限までの時間が指定のウィンドウ係数に達するまでの間、ログインスレッドはスリープ状態になります。この時間の経過後、チケットの更新が試行されます。

  • 型: double
  • デフォルト: 0.8
  • 重要度: 低
producer.sasl.kerberos.ticket.renew.jitter

更新時間に追加されたランダムジッターのパーセンテージ。

  • 型: double
  • デフォルト: 0.05
  • 重要度: 低
producer.sasl.kerberos.min.time.before.relogin

更新試行から次の更新試行までの、ログインスレッドのスリープ時間。

  • 型: long
  • デフォルト: 60000
  • 重要度: 低
producer.sasl.login.refresh.window.factor

認証情報の存続期間を基準とする指定のウィンドウ係数に達するまでの間、ログイン更新スレッドはスリープ状態になります。この時間の経過後、認証情報の更新が試行されます。指定可能な値は 0.5（50%）～ 1.0（100%）（1.0 を含む）の間です。値が指定されていない場合は、0.8（80%）のデフォルト値が使用されます。現時点では、OAUTHBEARER にのみ適用されます。

  • 型: double
  • デフォルト: 0.8
  • 指定可能な値: [0.5,...,1.0]
  • 重要度: 低
producer.sasl.login.refresh.window.jitter

ログイン更新スレッドのスリープ時間に追加される認証情報の存続期間を基準とする、ランダムジッターの最大量。指定可能な値は 0 ～ 0.25（25%）（0.25 を含む）の間です。値が指定されていない場合は、0.05（5%）のデフォルト値が使用されます。現時点では、OAUTHBEARER にのみ適用されます。

  • 型: double
  • デフォルト: 0.05
  • 指定可能な値: [0.0,...,0.25]
  • 重要度: 低
producer.sasl.login.refresh.min.period.seconds

ログイン更新スレッドが認証情報の更新まで待機する目的の最小時間（秒単位）。指定可能な値は 0 ～ 900（15 分）の間です。値が指定されていない場合は、60（1 分）のデフォルト値が使用されます。この値と sasl.login.refresh.buffer.seconds の合計が認証情報の残りの存続期間を超える場合は、その両方が無視されます。現時点では、OAUTHBEARER にのみ適用されます。

  • 型: short
  • デフォルト: 60
  • 指定可能な値: [0,...,900]
  • 重要度: 低
producer.sasl.login.refresh.buffer.seconds

認証情報の更新時に維持される認証情報の有効期限までのバッファ時間の長さ（秒単位）。バッファの秒数よりも有効期限に近い時点で、異なる形で更新が発生した場合、その更新は繰り上げられ、バッファ時間の長さの分だけ可能な限り維持されます。指定可能な値は 0 ～ 3600（1 時間）の間です。値が指定されていない場合は、300（5 分）のデフォルト値が使用されます。この値と sasl.login.refresh.min.period.seconds の合計が認証情報の残りの存続期間を超える場合は、その両方が無視されます。現時点では、OAUTHBEARER にのみ適用されます。

  • 型: short
  • デフォルト: 300
  • 指定可能な値: [0、…、3600]
  • 重要度: 低
producer.sasl.mechanism

クライアントの接続に使用される SASL メカニズム。セキュリティプロバイダーを利用できるものであれば、どのメカニズムでも構いません。GSSAPI がデフォルトのメカニズムです。

  • 型: string
  • デフォルト: GSSAPI
  • 重要度: 中
producer.sasl.jaas.config

JAAS の構成ファイルで使用される形式で記述された、SASL 接続の JAAS ログインコンテキストパラメーター。JAAS 構成ファイル形式については、<a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/LoginConfigFile.html">こちら</a>の説明を参照してください。値の形式は「<code>loginModuleClass controlFlag (optionName=optionValue)*;</code>」です。ブローカーの場合は、小文字のリスナープレフィックスと SASL メカニズム名を構成の先頭に付ける必要があります。たとえば、listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule とする必要があります。

  • 型: password
  • デフォルト: null
  • 重要度: 中
producer.sasl.client.callback.handler.class

AuthenticateCallbackHandler インターフェイスを実装する SASL クライアントコールバックハンドラークラスの完全修飾名。

  • 型: class
  • デフォルト: null
  • 重要度: 中
producer.sasl.login.callback.handler.class

AuthenticateCallbackHandler インターフェイスを実装する SASL ログインコールバックハンドラークラスの完全修飾名です。ブローカーの場合は、小文字のリスナープレフィックスと SASL メカニズム名をログインコールバックハンドラー構成の先頭に付ける必要があります。たとえば、listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler とします。

  • 型: class
  • デフォルト: null
  • 重要度: 中
producer.sasl.login.class

Login インターフェイスを実装するクラスの完全修飾名。ブローカーの場合は、小文字のリスナープレフィックスと SASL メカニズム名をログイン構成の先頭に付ける必要があります。たとえば、listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin とします。

  • 型: class
  • デフォルト: null
  • 重要度: 中

ライセンス

ちなみに

Confluent Platform のライセンス情報全般については、「Confluent Platform ライセンス」を参照してください。

confluent.license

Confluent では、各契約者にライセンスキーを発行します。ライセンスキーは、コピーアンドペーストできる短いテキストです。Kafka-MQTT は、ライセンスキーがなくても 30 日間試用できます。すでにご契約されている場合は、詳細について Confluent サポートにお問い合わせください。

  • 型: string
  • デフォルト: ""
  • 重要度: 高
confluent.topic.bootstrap.servers

ライセンス供与に使用される Kafka クラスターとの初期接続を確立するために使用するホストとポートのペアのリスト。初期接続で、クラスター内のすべてのサーバーが検出されます。このリストは、host1:port1,host2:port2,… という形式にする必要があります。これらのサーバーは、初期接続ですべてのクラスターメンバーシップを検出するためにのみ使用されます。これは動的に変わる可能性があるので、このリストにすべてのサーバーセットを含める必要はありません（ただし、サーバーの障害に備えて、複数指定しておくこともできます）。

  • 型: list
  • デフォルト: bootstrap.servers プロパティの値
  • 重要度: 高
confluent.topic

Confluent Platform の構成（ライセンス情報など）で使用される Kafka トピックの名前。

  • 型: string
  • デフォルト: _confluent-command
  • 重要度: 低
confluent.topic.replication.factor

Confluent Platform の構成（ライセンス情報など）で使用される Kafka のトピックのレプリケーション係数。これは、トピックがまだ存在しない場合にのみ使用されます。デフォルトの 3 は、本稼働環境での使用に適しています。ブローカー数を 3 未満の開発環境で使用している場合は、そのブローカー数をこのプロパティに設定する必要があります（多くの場合、1 です）。

  • 型: int
  • デフォルト: 3
  • 重要度: 低