RHEL での AMQ Streams の使用

Apache Kafka と ZooKeeper には別々のディスクを使用します。

Kafka は、複数のディスクまたはボリュームのデータストレージ設定である JBOD(Just a Bunch of Disks)ストレージをサポートします。JBOD は、Kafka ブローカーのデータストレージを増やします。また、パフォーマンスを向上することもできます。

ソリッドステートドライブ (SSD) は必須ではありませんが、複数のトピックに対してデータが非同期的に送受信される大規模なクラスターで Kafka のパフォーマンスを向上させることができます。SSD は、高速で低レイテンシーのデータアクセスが必要な ZooKeeper で特に有効です。

XFS ファイルシステムを使用するようにストレージシステムを設定することが推奨されます。AMQ Streams は ext4 ファイルシステムとも互換性がありますが、最適な結果を得るには追加の設定が必要になることがあります。

JAAS は個別の設定ファイルを使用して設定されます。JAAS 設定ファイルを ZooKeeper 設定と同じディレクトリー (/opt/kafka/config/) に置くことが推奨されます。推奨されるファイル名は zookeeper-jaas.conf です。複数のノードで ZooKeeper クラスターを使用する場合は、JAAS 設定ファイルをすべてのクラスターノードで作成する必要があります。

JAAS はコンテキストを使用して設定されます。サーバーとクライアントなどの個別の部分は、常に別の コンテキスト で設定されます。コンテキストは 設定 オプションで、以下の形式となっています。

ContextName {
       param1
       param2;
};

SASL 認証は、サーバー間通信 (Zoo Keeper インスタンス間の通信) とクライアント間通信 (Kafka と Zoo Keeper 間の通信) に対して別々に構成されます。サーバー間の認証は、複数のノードを持つ ZooKeeper クラスターにのみ関連します。

DIGEST-MD5 SASL メカニズムを使用する場合、認証サーバーの設定に QuorumServer コンテキストが使用されます。暗号化されていない形式で、接続できるすべてのユーザー名とパスワードが含まれている必要があります。2 つ目のコンテキスト QuorumLearner は、ZooKeeper に組み込まれるクライアント用に設定する必要があります。また、暗号化されていない形式のパスワードも含まれます。DIGEST-MD5 メカニズムの JAAS 設定ファイルの例は、以下を参照してください。

QuorumServer {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       user_zookeeper="123456";
};

QuorumLearner {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       username="zookeeper"
       password="123456";
};

JAAS 設定ファイルの他に、以下のオプションを指定して、通常の ZooKeeper 設定ファイルでサーバー間の認証を有効にする必要があります。

quorum.auth.enableSasl=true
quorum.auth.learnerRequireSasl=true
quorum.auth.serverRequireSasl=true
quorum.auth.learner.loginContext=QuorumLearner
quorum.auth.server.loginContext=QuorumServer
quorum.cnxn.threads.size=20

KAFKA_OPTS 環境変数を使用して、JAAS 設定ファイルを Java プロパティーとして ZooKeeper サーバーに渡します。

su - kafka
export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/zookeeper-jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

サーバー間の認証の詳細は、ZooKeeper Wiki を参照してください。

JAAS 設定ファイルにサーバーコンテキストを追加して、クライアント/サーバー間の認証を設定します。DIGEST-MD5 メカニズムの場合は、すべてのユーザー名とパスワードを設定します。

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    user_super="123456"
    user_kafka="123456"
    user_someoneelse="123456";
};

JAAS コンテキストの設定後、以下の行を追加して ZooKeeper 設定ファイルでクライアント/サーバー間の認証を有効にします。

requireClientAuthScheme=sasl
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
authProvider.2=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
authProvider.3=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

ZooKeeper クラスターの一部であるすべてのサーバーに authProvider.<ID> プロパティーを追加する必要があります。

KAFKA_OPTS 環境変数を使用して、JAAS 設定ファイルを Java プロパティーとして ZooKeeper サーバーに渡します。

su - kafka
export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/zookeeper-jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

Kafka ブローカーでの ZooKeeper 認証の設定に関する詳細は、「ZooKeeper の認証」 を参照してください。

この手順では、ZooKeeper クラスターのノード間で SASL DIGEST-MD5 メカニズムを使用した認証を有効にする方法を説明します。

この手順では、ZooKeeper クライアントと ZooKeeper との間で SASL DIGEST-MD5 メカニズムを使用した認証を有効にする方法を説明します。

ZooKeeper 接続の SASL 認証は JAAS 設定ファイルで設定する必要があります。デフォルトでは、Kafka は ZooKeeper への接続用に Client という名前の JAAS コンテキストを使用します。Clientコンテキストは /opt/kafka/config/jass.conf ファイルで設定する必要があります。以下の例のように、コンテキストでは PLAIN SASL 認証を有効にする必要があります。

Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="123456";
};

この手順では、ZooKeeper に接続する際に SASL DIGEST-MD5 メカニズムを使用した認証を有効にする方法を説明します。

AclAuthorizer を含む authorizer プラグインは authorizer.class.name プロパティーを使用して有効にします。

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

選択したオーソライザーには完全修飾名が必要です。AclAuthorizer の場合、完全修飾名は kafka.security.auth.SimpleAclAuthorizer です。

super.users=User:admin,User:operator

この手順では、Kafka ブローカーでの承認用に AclAuthorizer プラグインを有効にする方法を説明します。

AclAuthorizer は、ユーザーが実行できる/できない操作を記述するルールのセットを定義するアクセス制御リスト (ACL) を使用します。

この手順では、Kafka ブローカーで AclAuthorizer プラグインを使用する場合に、ACL ルールを追加する方法を説明します。

ルールは kafka-acls.sh ユーティリティーを使用して追加され、ZooKeeper に保存されます。

この手順では、Kafka ブローカーで AclAuthorizer プラグインを使用する場合に、既存の ACL ルールを一覧表示する方法を説明します。

ルールは、kafka-acls.sh ユーティリティーを使用してリストされます。

この手順では、Kafka ブローカーで AclAuthorizer プラグインを使用する場合に、ACL ルールを削除する方法を説明します。

ルールは kafka-acls.sh ユーティリティーを使用して削除されます。

ZooKeeper ACL ルールの適用は、config/server.properties Kafka 設定ファイルの zookeeper.set.acl プロパティーによって制御されます。

プロパティーはデフォルトで無効になっていて、true に設定することにより有効になります。

zookeeper.set.acl=true

ACL ルールが有効になっている場合、ZooKeeper で znode が作成されると、作成した Kafka ユーザーのみが変更または削除できます。その他のすべてのユーザーには読み取り専用アクセスがあります。

Kafka は、新しく作成された ZooKeeper znodes に対してのみ ACL ルールを設定します。ACL がクラスターの最初の起動後にのみ有効である場合、zookeeper-security-migration.sh ツールは既存のすべての znodes に ACL を設定できます。

しかし、ZooKeeper は Kafka を使用して送受信されたレコードを保存しません。ZooKeeper に保存されるデータは機密ではないと想定されます。

データが機密として考慮される場合 (たとえば、トピック名にカスタマー ID が含まれるなど)、保護に使用できる唯一のオプションは、ネットワークレベルで ZooKeeper を分離し、Kafka ブローカーにのみアクセスを許可することです。

この手順では、新しい Kafka クラスターの Kafka 設定で ZooKeeper ACL を有効にする方法を説明します。この手順は、Kafka クラスターの最初の起動前にのみ使用してください。すでに実行中のクラスターで ZooKeeper ACL を有効にする場合は、「既存の Kafka クラスターでの ZooKeeper ACL の有効化」 を参照してください。

この手順では、稼働している Kafka クラスターの Kafka 設定で ZooKeeper ACL を有効にする方法を説明します。zookeeper-security-migration.sh ツールを使用して、既存のすべての znodes に ZooKeeper の ACL を設定します。zookeeper-security-migration.sh は AMQ Streams の一部として利用でき、bin ディレクトリーにあります。

Kafka ブローカーの暗号化および認証は、リスナーごとに設定されます。Kafka リスナーの設定に関する詳細は、「「リスナー」」 を参照してください。

Kafka ブローカーの各リスナーは、独自のセキュリティープロトコルで設定されます。設定プロパティー listener.security.protocol.map は、どのリスナーがどのセキュリティープロトコルを使用するかを定義します。各リスナー名がセキュリティープロトコルにマッピングされます。サポートされるセキュリティープロトコルは次のとおりです。

以下の listeners 設定の場合、

listeners=INT1://:9092,INT2://:9093,REPLICATION://:9094

listener.security.protocol.map は以下のようになります。

listener.security.protocol.map=INT1:SASL_PLAINTEXT,INT2:SASL_SSL,REPLICATION:SSL

これにより、リスナー INT1 は暗号化されていない接続および SASL 認証を使用し、リスナー INT2 は暗号化された接続および SASL 認証を使用し、REPLICATION インターフェイスは TLS による暗号化 (TLS クライアント認証が使用される可能性があり) を使用するように設定されます。同じセキュリティープロトコルを複数回使用できます。以下は、有効な設定の例です。

listener.security.protocol.map=INT1:SSL,INT2:SSL,REPLICATION:SSL

このような設定は、すべてのインターフェイスに TLS 暗号化および TLS 認証を使用します。以下の章では、TLS および SASL の設定方法について詳しく説明します。

Kafka は、Kafka クライアントとの通信を暗号化するために TLS をサポートします。

TLS による暗号化およびサーバー認証を使用するには、秘密鍵と公開鍵が含まれるキーストアを提供する必要があります。これは通常、Java Keystore (JKS) 形式のファイルを使用して行われます。このファイルのパスは、ssl.keystore.location プロパティーに設定されます。ssl.keystore.password プロパティーを使用して、キーストアを保護するパスワードを設定する必要があります。以下に例を示します。

ssl.keystore.location=/path/to/keystore/server-1.jks
ssl.keystore.password=123456

秘密鍵を保護するために、追加のパスワードが使用されることがあります。このようなパスワードは、ssl.key.password プロパティーを使用して設定できます。

Kafka は、認証局によって署名された鍵と自己署名の鍵を使用できます。認証局が署名する鍵を使用することが、常に推奨される方法です。クライアントが接続している Kafka ブローカーのアイデンティティーを検証できるようにするには、証明書に Common Name (CN) または Subject Alternative Names (SAN) としてアドバタイズされたホスト名が常に含まれる必要があります。

異なるリスナーに異なる SSL 設定を使用できます。ssl. で始まるすべてのオプションの前に listener.name.<NameOfTheListener>. を付けることができます。この場合、リスナーの名前は常に小文字である必要があります。これにより、その特定のリスナーのデフォルトの SSL 設定が上書きされます。以下の例は、異なるリスナーに異なる SSL 設定を使用する方法を示しています。

listeners=INT1://:9092,INT2://:9093,REPLICATION://:9094
listener.security.protocol.map=INT1:SSL,INT2:SSL,REPLICATION:SSL

# Default configuration - will be used for listeners INT1 and INT2
ssl.keystore.location=/path/to/keystore/server-1.jks
ssl.keystore.password=123456

# Different configuration for listener REPLICATION
listener.name.replication.ssl.keystore.location=/path/to/keystore/server-1.jks
listener.name.replication.ssl.keystore.password=123456

サポートされる Kafka ブローカー設定オプションの完全リストは、「付録A ブローカー設定パラメーター」を参照してください。

この手順では、Kafka ブローカーで暗号化を有効にする方法を説明します。

認証には、以下を使用できます。

ssl.truststore.location=/path/to/keystore/server-1.jks
ssl.truststore.password=123456

KAFKA_OPTS="-Djava.security.auth.login.config=/path/to/my/jaas.config"; bin/kafka-server-start.sh

sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512

sasl.mechanism.inter.broker.protocol=PLAIN

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    user_admin="123456"
    user_user1="123456"
    user_user2="123456";
};

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="123456"
    user_admin="123456"
    user_user1="123456"
    user_user2="123456";
};

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required;
};

sasl.enabled.mechanisms=SCRAM-SHA-256,SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'SCRAM-SHA-256=[password=123456],SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name user1

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name user1

KafkaServer {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_server.keytab"
    principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
};

sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.kerberos.service.name=kafka

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    user_admin="123456"
    user_user1="123456"
    user_user2="123456";

    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_server.keytab"
    principal="kafka/kafka1.hostname.com@EXAMPLE.COM";

    org.apache.kafka.common.security.scram.ScramLoginModule required;
};

この手順では、Kafka ブローカーで TLS クライアント認証を有効にする方法を説明します。

この手順では、Kafka ブローカーで SASL PLAIN 認証を有効にする方法を説明します。

この手順では、Kafka ブローカーで SASL SCRAM 認証を有効にする方法を説明します。

この手順では、SASL SCRAM を使用した認証に新しいユーザーを追加する方法を説明します。

この手順では、SASL SCRAM 認証を使用する場合にユーザーを削除する方法を説明します。

Kafka SASL OAUTHBEARER メカニズムは、Kafka ブローカーで認証されたセッションを確立するために使用されます。

Kafka クライアントは、形式がアクセストークンであるクレデンシャルの交換に SASL OAUTHBEARER メカニズムを使用して Kafka ブローカーでセッションを開始します。

Kafka ブローカーおよびクライアントは、OAuth 2.0 を使用するように設定する必要があります。

OAuth 2.0 の Kafka ブローカー設定には、以下が関係します。

sasl.enabled.mechanisms=OAUTHBEARER 1
listeners=CLIENT://0.0.0.0:9092 2
listener.security.protocol.map=CLIENT:SASL_PLAINTEXT 3
listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER 4
sasl.mechanism.inter.broker.protocol=OAUTHBEARER 5
inter.broker.listener.name=CLIENT 6
listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler 7
listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ 8
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \ 9
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \ 10
  oauth.username.claim="preferred_username"  \ 11
  oauth.client.id="kafka-broker" \ 12
  oauth.client.secret="kafka-secret" \ 13
  oauth.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/token" ; 14
listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler 15
listener.name.client.oauthbearer.connections.max.reauth.ms=3600000 16

sasl.enabled.mechanisms=
listeners=REPLICATION://kafka:9091,CLIENT://kafka:9092 1
listener.security.protocol.map=REPLICATION:SSL,CLIENT:SASL_PLAINTEXT 2
listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER
inter.broker.listener.name=REPLICATION
listener.name.replication.ssl.keystore.password=KEYSTORE-PASSWORD 3
listener.name.replication.ssl.truststore.password=TRUSTSTORE-PASSWORD
listener.name.replication.ssl.keystore.type=JKS
listener.name.replication.ssl.truststore.type=JKS
listener.name.replication.ssl.endpoint.identification.algorithm=HTTPS 4
listener.name.replication.ssl.secure.random.implementation=SHA1PRNG 5
listener.name.replication.ssl.keystore.location=PATH-TO-KEYSTORE 6
listener.name.replication.ssl.truststore.location=PATH-TO-TRUSTSTORE 7
listener.name.replication.ssl.client.auth=required 8
listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \
  oauth.username.claim="preferred_username" ; 9

listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \ 1
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \ 2
  oauth.jwks.refresh.seconds="300" \ 3
  oauth.jwks.refresh.min.pause.seconds="1" \ 4
  oauth.jwks.expiry.seconds="360" \ 5
  oauth.username.claim="preferred_username" \ 6
  oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \ 7
  oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \ 8
  oauth.ssl.truststore.type="PKCS12" ; 9

listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.introspection.endpoint.uri="https://AUTH-SERVER-ADDRESS/introspection" \ 1
  oauth.client.id="kafka-broker" \ 2
  oauth.client.secret="kafka-broker-secret" \ 3
  oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \ 4
  oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \ 5
  oauth.ssl.truststore.type="PKCS12" \ 6
  oauth.username.claim="preferred_username" ; 7

AMQ Streams の OAuth 2.0 認証に使用される Kafka SASL OAUTHBEARER メカニズムは、再認証 メカニズムと呼ばれる Kafka 機能をサポートします。

リスナー設定で再認証メカニズムが有効になっている場合、アクセストークンの期限が切れるとブローカーの認証されたセッションが期限切れになります。その後、クライアントは接続を切断せずに新しい有効なアクセストークンをブローカーに送信し、既存のセッションを再認証する必要があります。

トークンの検証に成功すると、既存の接続を使用して新しいクライアントセッションが開始されます。クライアントが再認証に失敗した場合、さらにメッセージを送受信しようとすると、ブローカーは接続を閉じます。ブローカーで再認証メカニズムが有効になっていると、Kafka クライアントライブラリー 2.2 以降を使用する Java クライアントが自動的に再認証されます。

Kafka server.properties ファイルで Kafka ブローカーのセッションの再認証を有効にします。OAUTHBEARER を SASL メカニズムとして有効にした TLS リスナーに connections.max.reauth.ms プロパティーを設定します。

リスナーごとにセッションの再認証を指定できます。以下に例を示します。

listener.name.client.oauthbearer.connections.max.reauth.ms=3600000

セッションの再認証は、高速のローカル JWT とイントロスペクションエンドポイントの両タイプのトークン検証でサポートされます。設定例については、「Kafka ブローカーの OAuth 2.0 サポートの設定」 を参照してください。

Kafka バージョン 2.2 に追加された再認証メカニズムの詳細は、「KIP-368」を参照してください。

Kafka クライアントは以下のいずれかで設定されます。

クレデンシャルは Kafka ブローカーに送信されることはありません。アクセストークンは、Kafka ブローカーに送信される唯一の情報です。クライアントによるアクセストークンの取得後、承認サーバーと通信する必要はありません。

クライアント ID とシークレットを使用した認証が最も簡単です。有効期間の長いアクセストークンまたは更新トークンを使用すると、承認サーバーツールに追加の依存関係があるため、より複雑になります。

Kafka クライアントが直接アクセストークンで設定されていない場合、クライアントは承認サーバーと通信して Kafka セッションの開始中にアクセストークンのクレデンシャルを交換します。Kafka クライアントは以下のいずれかを交換します。

ここでは、Kafka セッションの開始時における Kafka クライアント、Kafka ブローカー、および承認ブローカー間の通信フローを説明および可視化します。フローは、クライアントとサーバーの設定によって異なります。

Kafka クライアントがアクセストークンをクレデンシャルとして Kafka ブローカーに送信する場合、トークンを検証する必要があります。

使用する承認サーバーや利用可能な設定オプションによっては、以下の使用が適している場合があります。

高速のローカルトークン検証を使用するには、トークンでの署名検証に使用される公開証明書のある JWKS エンドポイントを提供する承認サーバーが必要になります。

この他に、承認サーバーで OAuth 2.0 のイントロスペクションエンドポイントを使用することもできます。新しい Kafka ブローカー接続が確立されるたびに、ブローカーはクライアントから受け取ったアクセストークンを承認サーバーに渡し、応答を確認してトークンが有効であるかどうかを確認します。

Kafka クライアントのクレデンシャルは以下に対して設定することもできます。

OAuth 2.0 は、Kafka クライアントと AMQ Streams コンポーネントとの対話に使用されます。

AMQ Streams に OAuth 2.0 を使用するには、以下を行う必要があります。

AMQ Streams の OAuth 2.0 での承認では、Red Hat Single Sign-On サーバーの Authorization Services REST エンドポイントを使用して、Red Hat Single Sign-On を使用するトークンベースの認証が拡張されます。これは、定義されたセキュリティーポリシーを特定のユーザーに適用し、そのユーザーの異なるリソースに付与されたパーミッションの一覧を提供します。ポリシーはロールとグループを使用して、パーミッションをユーザーと照合します。OAuth 2.0 の承認では、Red Hat Single Sign-On の Authorization Services から受信した、ユーザーに付与された権限のリストを基にして、権限がローカルで強制されます。

この手順では、Red Hat Single Sign-On の Authorization Services を使用して、OAuth 2.0 承認を使用するように Kafka ブローカーを設定する方法を説明します。

通常、グループは組織の部門または地理的な場所を基にしてユーザーを照合するために使用されます。また、ロールは職務を基にしてユーザーを照合するために使用されます。

Red Hat Single Sign-On を使用すると、ユーザーおよびグループを LDAP で保存できますが、クライアントおよびロールは LDAP で保存できません。ユーザーデータへのアクセスとストレージを考慮して、承認ポリシーの設定方法を選択する必要がある場合があります。

OPA と AMQ Streams を統合する前に、粒度の細かいアクセス制御を提供するポリシーの定義方法を検討してください。

Kafka クラスター、コンシューマーグループ、およびトピックのアクセス制御を定義できます。たとえば、プロデューサークライアントから特定のブローカートピックへの書き込みアクセスを許可する承認ポリシーを定義できます。

このポリシーでは、以下の項目を指定することができます。

許可と拒否の決定がポリシーに書き込まれ、提供された要求とクライアント識別データに基づいて応答が提供されます。

この例では、プロデューサークライアントはトピックへの書き込みが許可されるポリシーを満たす必要があります。

Kafka が OPA ポリシーエンジンにアクセスしてアクセス制御ポリシーをクエリーできるようにするには、Kafka server.properties ファイルでカスタム OPA オーソライザープラグイン (kafka-authorizer-opa-VERSION.jar) を設定します。

クライアントがリクエストを行うと、OPA ポリシーエンジンは、指定された URL アドレスと REST エンドポイントを使用してプラグインによってクエリーされます。これは、定義されたポリシーの名前でなければなりません。

プラグインは、ポリシーに対してチェックされる JSON 形式で、クライアント要求の詳細 (ユーザープリンシパル、操作、およびリソース) を提供します。詳細には、クライアントの一意のアイデンティティーが含まれます。たとえば、TLS 認証が使用される場合にクライアント証明書からの識別名を取ります。

OPA はデータを使用して、リクエストを許可または拒否するためにプラグインに true または false のいずれかの応答を提供します。

この手順では、OPA 承認を使用するように Kafka ブローカーを設定する方法を説明します。

OPA を設定して、LDAP データソースからユーザー情報を読み込むことができます。

Kafka ブローカーロギングは、複数の 各ブローカーのブローカーロガー によって提供されます。ブローカーを再起動することなく、ブローカーロガーのロギングレベルを動的に変更できます。ログで返される詳細度レベルを上げると (たとえば、INFO から DEBUG に変更)、Kafka クラスターでパフォーマンスの問題を調査するのに役立ちます。

ブローカーロガーは、デフォルトのロギングレベルに動的にリセットすることもできます。

ブローカーロガーのリセット

kafka-configs.sh ツールを使用して、1 つ以上のブローカーロガーをデフォルトのロギングレベルにリセットできます。--alter および --delete-config オプションを使用して、各ブローカーロガーを二重引用符のコンマ区切りリストとして指定します。

/opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --delete-config "LOGGER-ONE,LOGGER-TWO" --entity-type broker-loggers --entity-name BROKER-ID

接続およびシリアライザープロパティーはすべてのプロデューサーに必要です。通常、追跡用のクライアント ID を追加し、プロデューサーで圧縮してリクエストのバッチサイズを減らすことが推奨されます。

基本的なプロデューサー設定は、以下のようになります。

# ...
bootstrap.servers=localhost:9092 1
key.serializer=org.apache.kafka.common.serialization.StringSerializer 2
value.serializer=org.apache.kafka.common.serialization.StringSerializer 3
client.id=my-client 4
compression.type=gzip 5
# ...

メッセージ配信の完了通知を使用して、データの持続性を適用し、メッセージが失われる可能性を最小限に抑えることができます。

# ...
acks=all 1
# ...

完了通知がプロデューサーに送信される前にメッセージをログに追加する必要のあるブローカーの数は、トピックの min.insync.replicas 設定によって決定されます。最初に、トピックレプリケーション係数を 3 にし、他のブローカーの In-Sync レプリカを 2 にするのが一般的です。この設定では、単一のブローカーが利用できない場合でもプロデューサーは影響を受けません。2 番目のブローカーが利用できなくなると、プロデューサーは完了通知を受信せず、それ以上のメッセージを生成できなくなります。

# ...
min.insync.replicas=2 1
# ...

メッセージは 1 度だけ配信されるため、べき等プロデューサーは重複を回避します。障害発生時でも配信の順序が維持されるように、ID とシーケンス番号がメッセージに割り当てられます。データの一貫性を保つために acks=all を使用している場合は、順序付けられた配信に冪等性を有効にすることが妥当です。

# ...
enable.idempotence=true 1
max.in.flight.requests.per.connection=5 2
acks=all 3
retries=2147483647 4
# ...

パフォーマンスコストが原因で acks=all と idempotency を使用していない場合には、インフライト (承認されていない) リクエストの数を 1 に設定して、順序を保持します。そうしないと、Message-A が失敗し、Message-B がブローカーに書き込まれた後にのみ成功する可能性があります。

# ...
enable.idempotence=false 1
max.in.flight.requests.per.connection=1 2
retries=2147483647
# ...

べき等は、1 つのパーティションへの書き込みを 1 回だけ行う場合に便利です。トランザクションをべき等と使用すると、複数のパーティション全体で 1 度だけ書き込みを行うことができます。

トランザクションは、同じトランザクション ID を使用するメッセージが 1 度作成され、すべて がそれぞれのログに書き込まれるか、何も 書き込まれないかのどちらかになることを保証します。

# ...
enable.idempotence=true
max.in.flight.requests.per.connection=5
acks=all
retries=2147483647
transactional.id=UNIQUE-ID 1
transaction.timeout.ms=900000 2
# ...

トランザクション保証を維持するには、transactional.id の選択が重要です。トランザクション ID は、一意なトピックパーティションセットに使用する必要があります。たとえば、トピックパーティション名からトランザクション ID への外部マッピングを使用したり、競合を回避する関数を使用してトピックパーティション名からトランザクション IDを算出したりすると、これを実現できます。

通常、システムの要件は、指定のレイテンシー内であるメッセージの割合に対して、特定のスループットのターゲットを達成することです。たとえば、95 ％ のメッセージが 2 秒以内に完了確認される、1 秒あたり 500,000 個のメッセージをターゲットとします。

プロデューサーのメッセージングセマンティック (メッセージの順序付けと持続性) は、アプリケーションの要件によって定義される可能性があります。たとえば、アプリケーションが提供する重要なプロパティーや保証を壊さずに acks=0 または acks=1 を使用するオプションはありません。

ブローカーの再起動は、パーセンタイルの高いの統計に大きく影響します。たとえば、長期間では、99％ のレイテンシーはブローカーの再起動に関する動作によるものです。これは、ベンチマークを設計したり、本番環境のパフォーマンスで得られた数字を使ってベンチマークを行い、そのパフォーマンスの数字を比較したりする場合に検討する価値があります。

目的に応じて、Kafka はスループットとレイテンシーのプロデューサーパフォーマンスを調整するために多くの設定パラメーターと設定方法を提供します。

ここで、send() メソッドが返されます。そのため、send() がブロックされる時間は、以下によって決定されます。

バッチは、以下のいずれかが行われるまでキューに残ります。

バッチ処理とバッファーの設定を参照して、レイテンシーをブロックする send() の影響を軽減します。

# ...
linger.ms=100 1
batch.size=16384 2
buffer.memory=33554432 3
# ...

また、カスタムパーティションを作成してデフォルトを置き換えることで、メッセージを指定のパーティションに転送することもできます。

# ...
delivery.timeout.ms=120000 1
partitioner.class=my-custom-partitioner 2

# ...

接続およびデシリアライザープロパティーはすべてのコンシューマーに必要です。通常、追跡用にクライアント ID を追加することが推奨されます。

コンシューマー設定では、後続の設定に関係なく、以下を行います。

# ...
bootstrap.servers=localhost:9092 1
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer  2
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer  3
client.id=my-client 4
group.id=my-group-id 5
# ...

コンシューマーグループは通常、特定のトピックから複数のプロデューサーで生成される大規模なデータストリームを共有するために使用されます。コンシューマーは group.id でグループ化され、メッセージをメンバー全体に分散できます。

コンシューマーグループは、特定のトピックから 1 つまたは複数のプロデューサーによって生成される、典型的な大量のデータストリームを共有します。group.id プロパティーが同じコンシューマーは同じグループにあります。グループ内のコンシューマーの 1 つがリーダーを選択し、パーティションをグループのコンシューマーにどのように割り当てるかを決定します。各パーティションは 1 つのコンシューマーにのみ割り当てることができます。

コンシューマーの数がパーティション数だけない場合には、同じ group.id を持つコンシューマーインスタンスを追加して、データ消費をスケーリングできます。コンシューマーをグループに追加して、パーティションの数より多くしても、スループットは改善されませんが、コンシューマーが機能しなくなったときに予備のコンシューマーを使用できます。より少ないコンシューマーでスループットの目標を達成できれば、リソースを節約できます。

同じコンシューマーグループのコンシューマーは、オフセットコミットとハートビートを同じブローカーに送信します。グループのコンシューマーの数が多いほど、ブローカーのリクエスト負荷が高くなります。

# ...
group.id=my-group-id 1
# ...

Kafka ブローカーは、トピック、パーティション、およびオフセット位置のリストからメッセージを送信するようブローカーに要求するコンシューマーからフェッチリクエストを受け取ります。

コンシューマーは、ブローカーにコミットされたのと同じ順序でメッセージを単一のパーティションで監視します。つまり、Kafka は単一パーティションのメッセージ のみ 順序付けを保証します。逆に、コンシューマーが複数のパーティションからメッセージを消費している場合、コンシューマーによって監視される異なるパーティションのメッセージの順序は、必ずしも送信順序を反映しません。

1 つのトピックからメッセージを厳格に順序付ける場合は、コンシューマーごとに 1 つのパーティションを使用します。

クライアントアプリケーションが KafkaConsumer.poll() を呼び出すときに返されるメッセージの数を制御します。

fetch.max.wait.ms および fetch.min.bytes プロパティーを使用して、Kafka ブローカーからコンシューマーによって取得される最小データ量を増やします。時間ベースのバッチ処理は fetch.max.wait.ms を使用して設定され、サイズベースのバッチ処理は fetch.min.bytes を使用して設定されます。

コンシューマーまたはブローカーの CPU 使用率が高い場合、コンシューマーからのリクエストが多すぎる可能性があります。fetch.max.wait.ms プロパティーおよび fetch.min.bytes プロパティーを調整して、より大きなバッチで要求とメッセージが配信されるようにすることができます。より高い値に調整することでスループットが改善されますが、レイテンシーのコストが発生します。生成されるデータ量が少ない場合、より高い値に調整することもできます。

たとえば、fetch.max.wait.ms を 500ms に設定し、fetch.min.bytes を 16384 バイトに設定した場合、Kafka がコンシューマーからフェッチリクエストを受信すると、いずれかのしきい値に最初に到達した時点で応答されます。

逆に、fetch.max.wait.ms プロパティーおよび fetch.min.bytes プロパティーを調整して、エンドツーエンドのレイテンシーを改善できます。

# ...
fetch.max.wait.ms=500 1
fetch.min.bytes=16384 2
# ...

fetch.max.bytes プロパティーは、一度にブローカーから取得されるデータ量の上限をバイト単位で設定します。

max.partition.fetch.bytes は、各パーティションで返されるデータ量の上限をバイト単位で設定します。これは、max.message.bytes のブローカーまたはトピック設定に設定されたバイト数よりも大きくする必要があります。

クライアントが消費できるメモリーの最大量は、以下のように概算されます。

NUMBER-OF-BROKERS * fetch.max.bytes and NUMBER-OF-PARTITIONS * max.partition.fetch.bytes

メモリー使用量がこれに対応できる場合は、これら 2 つのプロパティーの値を増やすことができます。各リクエストでより多くのデータを許可すると、フェッチリクエストが少なくなるため、レイテンシーが向上されます。

# ...
fetch.max.bytes=52428800 1
max.partition.fetch.bytes=1048576 2
# ...

Kafka の 自動コミットメカニズム により、コンシューマーはメッセージのオフセットを自動的にコミットできます。有効にすると、コンシューマーはブローカーをポーリングして受信したオフセットを 5000ms 間隔でコミットします。

自動コミットのメカニズムは便利ですが、データ損失と重複のリスクが発生します。コンシューマーが多くのメッセージを取得および変換し、自動コミットの実行時にコンシューマーバッファーに処理されたメッセージがある状態でシステムがクラッシュすると、そのデータは失われます。メッセージの処理後、自動コミットの実行前にシステムがクラッシュした場合、リバランス後に別のコンシューマーインスタンスでデータが複製されます。

ブローカーへの次のポーリングの前またはコンシューマーが閉じられる前に、すべてのメッセージが処理された場合は、自動コミットによるデータの損失を回避できます。

データの損失や重複の可能性を最小限に抑えるには、enable.auto.commit を false に設定し、クライアントアプリケーションを開発して、オフセットのコミットをより詳細に制御できるようにします。または、auto.commit.interval.ms を使用してコミットの間隔を減らすことができます。

# ...
enable.auto.commit=false 1
# ...

enable.auto.commit を false に設定すると、すべての 処理が実行され、メッセージが消費された後にオフセットをコミットできます。たとえば、Kafka commitSync および commitAsync コミット API を呼び出すようにアプリケーションを設定できます。

commitSync API は、ポーリングから返されるメッセージバッチのオフセットをコミットします。バッチのメッセージすべての処理が完了したら API を呼び出します。commitSync API を使用する場合、アプリケーションはバッチの最後のオフセットがコミットされるまで新しいメッセージをポーリングしません。これがスループットに悪影響する場合は、コミットする頻度が低いか、commitAsync API を使用できます。commitAsync API はブローカーがコミットリクエストに応答するまで待機しませんが、リバランス時にさらに重複が発生するリスクがあります。一般的な方法として、両方のコミット API をアプリケーションで組み合わせ、コンシューマーをシャットダウンまたはリバランスの直前に commitSync API を使用し、最終コミットが正常に実行されるようにします。

# ...
enable.auto.commit=false
isolation.level=read_committed 1
# ...

session.timeout.ms および heartbeat.interval.ms プロパティーを使用して、コンシューマーグループ内のコンシューマー障害をチェックし、復旧するのにかかった時間を設定します。

session.timeout.ms プロパティーは、コンシューマーグループ内のコンシューマーがブローカーとのコンタクトを絶った場合に、非アクティブとみなされ、グループ内のアクティブなコンシューマー間で リバランス が発生するまでの最大時間をミリ秒単位で指定します。グループのリバランス時に、パーティションはグループのメンバーに再割り当てされます。

heartbeat.interval.ms プロパティーは、コンシューマーがアクティブで接続されていることを示す、コンシューマーグループコーディネーターへの ハートビート チェックの間隔をミリ秒単位で指定します。通常、ハートビートの間隔はセッションタイムアウトの間隔の 3 分の 2 にする必要があります。

session.timeout.ms プロパティーを低く設定すると、失敗したコンシューマーが先に検出され、リバランスがより迅速に実行されます。ただし、タイムアウトの値を低くしすぎて、ブローカーがハートビートを時間内に受信できず、不必要なリバランスがトリガーされることがないように気を付けてください。

ハートビートの間隔が短くなると、誤ってリバランスを行う可能性が低くなりますが、ハートビートを頻繁に行うとブローカーリソースのオーバーヘッドが増えます。

auto.offset.reset プロパティーを使用して、オフセットがコミットされていない場合にコンシューマーの動作を制御するか、コミットされたオフセットが有効でなくなったりします。

コンシューマーアプリケーションを初めてデプロイし、既存のトピックからメッセージを読み取る場合について考えてみましょう。これは group.id が初めて使用されるため、__consumer_offsets トピックには、このアプリケーションのオフセット情報は含まれません。新しいアプリケーションは、ログの始めからすべての既存メッセージの処理を開始するか、新しいメッセージのみ処理を開始できます。デフォルトのリセット値は、パーティションの最後に開始する latest で、一部のメッセージは見逃されることを意味します。データの損失は避けたいが、処理量を増やしたい場合は、auto.offset.reset を earliest に設定して、パーティションの先頭から開始します。

また、ブローカーに設定されたオフセットの保持期間 (offsets.retention.minutes) が終了したときにメッセージが失われるのを防ぐために、earliest オプションの使用も検討してください。コンシューマーグループまたはスタンドアロンコンシューマーが非アクティブで、保持期間中にオフセットをコミットしない場合、以前にコミットされたオフセットは __consumer_offsets から削除されます。

# ...
heartbeat.interval.ms=3000 1
session.timeout.ms=10000 2
auto.offset.reset=earliest 3
# ...

1 つのフェッチリクエストで返されるデータ量が大きい場合、コンシューマーが処理する前にタイムアウトが発生することがあります。この場合、max.partition.fetch.bytes を減らしたり、session.timeout.ms を増やすこともできます。

グループのアクティブなコンシューマー間で行うパーティションのリバランスは、以下にかかる時間です。

明らかに、このプロセスは特にコンシューマーグループクラスターのローリング再起動時に繰り返し発生するサービスのダウンタイムを増やします。

このような場合、静的メンバーシップ の概念を使用してリバランスの数を減らすことができます。リバランスによって、コンシューマーグループメンバー全体でトピックパーティションが割り当てられます。静的メンバーシップは永続性を使用し、セッションタイムアウト後の再起動時にコンシューマーインスタンスが認識されるようにします。

コンシューマーグループコーディネーターは、group.instance.id プロパティーを使用して指定される一意の ID を使用して新しいコンシューマーインスタンスを特定できます。再起動時には、コンシューマーには新しいメンバー ID が割り当てられますが、静的メンバーとして、同じインスタンス ID を使用し、同じトピックパーティションの割り当てが行われます。

コンシューマーアプリケーションが少なくとも max.poll.interval.ms ミリ秒毎にポーリングへの呼び出しを行わない場合、コンシューマーが失敗したと見なされ、リバランスが発生します。アプリケーションがポーリングから返されたすべてのレコードを時間内に処理できない場合は、max.poll.interval.ms プロパティーを使用してコンシューマーから新しいメッセージのポーリングの間隔をミリ秒単位で指定して、リバランスを回避することができます。または、max.poll.records プロパティーを使用して、コンシューマーバッファーから返されるレコードの数の上限を設定できます。これにより、アプリケーションは max.poll.interval.ms の制限内のより少ないレコードを処理できます。

# ...
group.instance.id=UNIQUE-ID 1
max.poll.interval.ms=300000 2
max.poll.records=500 3
# ...

トピックのスループットを向上させる主な方法は、そのトピックのパーティション数を増やすことです。これは、パーティションによってクラスター内のブローカー間でそのトピックの負荷が共有できるためです。ブローカーがすべて何らかのリソース (通常は I/O) によって制約されている場合、より多くのパーティションを使用してもスループットは向上しません。代わりに、クラスターにブローカーを追加する必要があります。

追加のブローカーをクラスターに追加する場合、AMQ Streams ではパーティションは自動的に割り当てられません。既存のブローカーから新しいブローカーに移動するパーティションを決定する必要があります。

すべてのブローカー間でパーティションが再分散されたら、各ブローカーのリソース使用率が低下するはずです。

クラスターからブローカーを削除する前に、そのブローカーにパーティションが割り当てられていないことを確認します。使用を停止するブローカーの各パーティションに対応する残りのブローカーを決定する必要があります。ブローカーに割り当てられたパーティションがない場合は、ブローカーを停止することができます。

再割り当て JSON ファイル には特定の構造があります。

{
  "version": 1,
  "partitions": [
    <PartitionObjects>
  ]
}

ここで <PartitionObjects> は、以下のようなコンマ区切りのオブジェクトリストになります。

{
  "topic": <TopicName>,
  "partition": <Partition>,
  "replicas": [ <AssignedBrokerIds> ],
  "log_dirs": [<LogDirs>]
}

"log_dirs" プロパティーはオプションで、パーティションを特定のログディレクトリーに移動するために使用されます。

以下は、トピック topic-a およびパーティション 4 をブローカー 2、4 および 7 に割り当て、トピック topic-b およびパーティション 2 をブローカー 1、5、および 7 に割り当てる、再割り当て JSON ファイルの例になります。

{
  "version": 1,
  "partitions": [
    {
      "topic": "topic-a",
      "partition": 4,
      "replicas": [2,4,7]
    },
    {
      "topic": "topic-b",
      "partition": 2,
      "replicas": [1,5,7]
    }
  ]
}

JSON に含まれていないパーティションは変更されません。

指定のトピックのセットのすべてのパーティションを、指定のブローカーのセットに割り当てる最も簡単な方法は、kafka-reassign-partitions.sh --generate コマンドを使用して再割り当て JSON ファイルを生成することです。

bin/kafka-reassign-partitions.sh --zookeeper <ZooKeeper> --topics-to-move-json-file <TopicsFile> --broker-list <BrokerList> --generate

<TopicsFile> は、移動するトピックをリストする JSON ファイルです。これには、以下の構造があります。

{
  "version": 1,
  "topics": [
    <TopicObjects>
  ]
}

ここで <TopicObjects> は、以下のようなコンマ区切りのオブジェクトリストになります。

{
  "topic": <TopicName>
}

たとえば、topic-a および topic-b のすべてのパーティションをブローカー 4 および 7 に移動する場合は、以下を実行します。

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topics-to-be-moved.json --broker-list 4,7 --generate

topics-to-be-moved.json のコンテンツがあります。

{
  "version": 1,
  "topics": [
    { "topic": "topic-a"},
    { "topic": "topic-b"}
  ]
}

特定のパーティションを移動したい場合は、再割り当て JSON ファイルを手動で作成できます。

以下の表は、Kafka サーバーに関する情報を報告するメトリクスの一部です。

以下の表は、リクエストに関する情報を報告するメトリクスの一部です。

次の表は、ロギングに関する情報を報告するメトリクスの選択を示しています。

次の表は、クラスターのコントローラーに関する情報を報告するメトリクスの選択を示しています。

レートまたは時間の単位を表すメトリクスは、Yammer メトリクスとして提供されます。Yammer メトリクスを使用する MBean のクラス名には、com.yammer.metrics というプレフィックスがつきます。

Yammer レートメトリクスには、要求を監視する以下の属性があります。

Yammer 時間メトリクスには、要求を監視するための以下の属性があります。

これらはプロデューサーレベルでのメトリクスです。

これらは、各ブローカーへの接続に関するプロデューサーレベルのメトリクスです。

これらは、プロデューサーがメッセージを送信しているトピックに関するトピックレベルのメトリクスです。

これらは、コンシューマーレベルのメトリクスです。

これらは、各ブローカーへの接続に関するコンシューマーレベルのメトリクスです。

これらは、コンシューマーグループに関するコンシューマーレベルのメトリクスです。

これらは、コンシューマーのフェッチャーに関するコンシューマーレベルのメトリクスです。

これらは、コンシューマーのフェッチャーに関するトピックレベルのメトリクスです。

これらは、コンシューマーのフェッチャーに関するパーティションレベルのメトリクスです。

これらは接続レベルのメトリクスです。

これらは、各ブローカーへの接続に関する接続レベルのメトリクスです。

これらは接続レベルのメトリクスです。

これらのメトリクスは、metrics.recording.level 設定パラメーターが info または debug の場合に収集されます。

タスクメトリクス

これらのメトリクスは、metrics.recording.level 設定パラメーターが debug のときに収集されます。

プロセッサーノードメトリクス

これらのメトリクスは、metrics.recording.level 設定パラメーターが debug のときに収集されます。

ステートストアメトリクス

これらのメトリクスは、metrics.recording.level 設定パラメーターが debug のときに収集されます。