ZooKeeper を使用した RHEL 上の Streams for Apache Kafka の使用

手順

1. クラスター内のすべての Kafka ブローカーの TLS 証明書を生成します。証明書には、Common Name または Subject Alternative Name にアドバタイズされたアドレスおよびブートストラップアドレスが必要です。

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • listener.security.protocol.map フィールドを変更して、TLS 暗号化を使用するリスナーに SSL プロトコルを指定します。
   • ssl.keystore.location オプションを、ブローカー証明書を持つ JKS キーストアへのパスに設定します。

   • ssl.keystore.password オプションを、キーストアの保護に使用したパスワードに設定します。

     以下に例を示します。

     Copy to Clipboard Toggle word wrap

     listeners=UNENCRYPTED://:9092,ENCRYPTED://:9093,REPLICATION://:9094
     listener.security.protocol.map=UNENCRYPTED:PLAINTEXT,ENCRYPTED:SSL,REPLICATION:PLAINTEXT
     ssl.keystore.location=/path/to/keystore/server-1.jks
     ssl.keystore.password=123456

3. Kafka ブローカーを (再) 起動します。

手順

1. ユーザー証明書の署名に使用される CA (認証局) の公開鍵を含む JKS (Java Keystore) トラストストアを準備します。

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • ssl.truststore.location プロパティーを使用して JKS トラストストアへのパスを指定します。
   • トラストストアがパスワードで保護される場合は、ssl.truststore.password プロパティーを使用してパスワードを設定します。

   • ssl.client.auth プロパティーを required に設定します。

     TLS クライアント認証の設定

     Copy to Clipboard Toggle word wrap

     ssl.truststore.location=/path/to/truststore.jks
     ssl.truststore.password=123456
     ssl.client.auth=required

3. Kafka ブローカーを (再) 起動します。

手順

1. /opt/kafka/config/jaas.conf JAAS 設定ファイルを編集または作成して、PlainLoginModule を有効にし、許可されるユーザー名とパスワードを指定します。

   このファイルがすべての Kafka ブローカーで同じであることを確認します。

   JAAS 設定

   Copy to Clipboard Toggle word wrap

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

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • listener.security.protocol.map プロパティーを使用して、特定のリスナーで SASL PLAIN 認証を有効にします。SASL_PLAINTEXT または SASL_SSL を指定します。

   • sasl.enabled.mechanisms プロパティーを PLAIN に設定します。

     SASL plain 設定

     Copy to Clipboard Toggle word wrap

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=PLAIN

3. KAFKA_OPTS 環境変数を使用して Kafka ブローカーを (再) 起動し、JAAS 設定を Kafka ブローカーに渡します。

   Copy to Clipboard Toggle word wrap

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

手順

1. /opt/kafka/config/jaas.conf JAAS 設定ファイルを編集または作成して、ScramLoginModule を有効にします。

   このファイルがすべての Kafka ブローカーで同じであることを確認します。

   JAAS 設定

   Copy to Clipboard Toggle word wrap

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

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • listener.security.protocol.map プロパティーを使用して、特定のリスナーで SASL SCRAM 認証を有効にします。SASL_PLAINTEXT または SASL_SSL を指定します。

   • sasl.enabled.mechanisms オプションを SCRAM-SHA-256 または SCRAM-SHA-512 に設定します。

     以下に例を示します。

     Copy to Clipboard Toggle word wrap

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=SCRAM-SHA-512

3. KAFKA_OPTS 環境変数を使用して Kafka ブローカーを (再) 起動し、JAAS 設定を Kafka ブローカーに渡します。

   Copy to Clipboard Toggle word wrap

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

手順

1. sasl.mechanism.inter.broker.protocol プロパティーを使用して、Kafka 設定でブローカー間 SASL メカニズムを指定します。

   ブローカー間 SASL メカニズム

   Copy to Clipboard Toggle word wrap

   sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

2. (オプション) SCRAM メカニズムを使用している場合は、SCRAM ユーザーを追加 して、Kafka クラスターに SCRAM 認証情報を登録します。

   これにより、Kafka クラスターの実行前に、認証用の認証情報がブートストラップ用に更新されるようになります。

3. username フィールドと password フィールドを使用して、KafkaServer JAAS コンテキストでのブローカー間通信用のユーザー名とパスワードを指定します。

   ブローカー間 JAAS コンテキスト

   Copy to Clipboard Toggle word wrap

   KafkaServer {
       org.apache.kafka.common.security.plain.ScramLoginModule required
       username="admin"
       password="123456"
       # ...
   };

1. サービスプリンシパル
2. Kafka ブローカー (Kerberos ログインを使用するため)
3. ZooKeeper (Kerberos ログインを使用するため)
4. プロデューサーおよびコンシューマークライアント (Kerberos 認証を使用して Kafka にアクセスするため)

1. Kerberos KDC を使用してサービスプリンシパルと、プリンシパルキーを保存するキータブを作成します。

   Kerberos プリンシパルのドメイン名が大文字であることを確認してください。

   以下に例を示します。

   • zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

     ZooKeeper サービスプリンシパルは、Kafka config/server.properties ファイルの zookeeper.connect 設定と同じホスト名にする必要があります。

     Copy to Clipboard Toggle word wrap

     zookeeper.connect=node1.example.redhat.com:2181

     ホスト名が異なる場合は localhost が使用されるため、認証に失敗します。

2. ホストにディレクトリーを作成し、キータブファイルを追加します。

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   /opt/kafka/krb5/zookeeper-node1.keytab
   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

3. kafka ユーザーがディレクトリーにアクセスできることを確認します。

   Copy to Clipboard Toggle word wrap

   chown kafka:kafka -R /opt/kafka/krb5

1. opt/kafka/config/jaas.conf ファイルを作成または変更して、ZooKeeper クライアントおよびサーバー操作をサポートします。

   Copy to Clipboard Toggle word wrap

   Client {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true 

   1

   
       storeKey=true 

   2

   
       useTicketCache=false 

   3

   
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 

   4

   
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 

   5

   
   };
   
   Server {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumServer {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumLearner {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   1

   true に設定し、キータブからプリンシパルキーを取得します。

   2

   true に設定し、プリンシパルキーを保存します。

   3

   true に設定し、チケットキャッシュから Ticket Granting Ticket (TGT) を取得します。

   4

   keyTab プロパティーは、Kerberos KDC からコピーされた keytab ファイルの場所を示します。その場所とファイルは、kafka ユーザーが読み取り可能になものにする必要があります。

   5

   principal プロパティーは、KDC ホストで作成された完全修飾プリンシパル名と一致するように設定され、その形式は SERVICE-NAME/FULLY-QUALIFIED-HOST-NAME@DOMAIN-NAME に従います。

2. opt/kafka/config/zookeeper.properties を編集して、更新された JAAS 設定を使用します。

   Copy to Clipboard Toggle word wrap

   # ...
   
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000 

   1

   
   kerberos.removeHostFromPrincipal=false 

   2

   
   kerberos.removeRealmFromPrincipal=false 

   3

   
   quorum.auth.enableSasl=true 

   4

   
   quorum.auth.learnerRequireSasl=true 

   5

   
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner 

   6

   
   quorum.auth.server.loginContext=QuorumServer
   quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 

   7

   
   quorum.cnxn.threads.size=20

   1

   ログイン更新の頻度をミリ秒単位で制御します。これは、チケットの更新間隔に合わせて調整できます。デフォルトは 1 時間です。

   2

   ホスト名がログインプリンシパル名の一部として使用されるかどうかを指定します。クラスターのすべてのノードで単一の keytab を使用する場合、これは true に設定されます。ただし、トラブルシューティングのために、各ブローカーホストに個別のキータブと完全修飾プリンシパルを生成することが推奨されます。

   3

   Kerberos ネゴシエーションのプリンシパル名からレルム名を削除するかどうかを制御します。この設定は、false にすることを推奨します。

   4

   ZooKeeper サーバーおよびクライアントの SASL 認証メカニズムを有効にします。

   5

   RequireSasl プロパティーは、マスター選出などのクォーラムイベントに SASL 認証を必要とするかどうかを制御します。

   6

   loginContext プロパティーは、指定されたコンポーネントの認証設定に使用される JAAS 設定のログインコンテキストの名前を識別します。loginContext 名は、opt/kafka/config/jaas.conf ファイルの関連セクションの名前に対応します。

   7

   識別に使用されるプリンシパル名を形成するために使用される命名規則を制御します。プレースホルダー _HOST は、実行時に server.1 プロパティーによって定義されたホスト名に自動的に解決されます。

3. JVM パラメーターで ZooKeeper を起動し、Kerberos ログイン設定を指定します。

   Copy to Clipboard Toggle word wrap

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

   デフォルトのサービス名 (zookeeper) を使用していない場合は、-Dzookeeper.sasl.client.username=NAME パラメーターを使用して名前を追加します。

   注記

   /etc/krb5.conf を場所として使用している場合は、ZooKeeper、Kafka、Kafka プロデューサーおよびコンシューマーの起動時に -Djava.security.krb5.conf=/etc/krb5.conf を指定する必要はありません。

1. opt/kafka/config/jaas.conf ファイルを以下の要素で修正します。

   Copy to Clipboard Toggle word wrap

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

2. Kafka クラスターの各ブローカーを設定するには、config/server.properties ファイルのリスナー設定を変更して、リスナーが SASL/GSSAPI ログインを使用するようにします。

   リスナーのセキュリティープロトコルのマップに SASL プロトコルを追加し、不要なプロトコルを削除します。

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 

   1

   
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 

   2

   
   # ..
   sasl.enabled.mechanisms=GSSAPI 

   3

   
   sasl.mechanism.inter.broker.protocol=GSSAPI 

   4

   
   sasl.kerberos.service.name=kafka 

   5

   
   ...

   1

   クライアントとの汎用通信用のセキュアなリスナー (通信用 TLS をサポート)、およびブローカー間通信用のレプリケーションリスナーの 2 つが設定されています。

   2

   TLS に対応しているリスナーのプロトコル名は SASL_PLAINTEXT になります。コネクターが TLS に対応していない場合、プロトコル名は SASL_PLAINTEXT になります。SSL が必要ない場合は、ssl.* プロパティーを削除できます。

   3

   Kerberos 認証における SASL メカニズムは GSSAPI になります。

   4

   ブローカー間通信の Kerberos 認証。

   5

   認証要求に使用するサービスの名前は、同じ Kerberos 設定を使用している他のサービスと区別するために指定されます。

3. Kafka ブローカーを起動し、JVM パラメーターを使用して Kerberos ログイン設定を指定します。

   Copy to Clipboard Toggle word wrap

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

   ブローカーおよび ZooKeeper クラスターが設定されていて、Kerberos ベース以外の認証システムで動作している場合は、ZooKeeper およびブローカークラスターを起動し、ログで設定エラーを確認できます。

   ブローカーおよび Zookeeper インスタンスを起動すると、Kerberos 認証用にクラスターが設定されました。

1. プロデューサーまたはコンシューマー設定ファイルに Kerberos 設定を追加します。

   以下に例を示します。

   /opt/kafka/config/producer.properties

   Copy to Clipboard Toggle word wrap

   # ...
   sasl.mechanism=GSSAPI 

   1

   
   security.protocol=SASL_PLAINTEXT 

   2

   
   sasl.kerberos.service.name=kafka 

   3

   
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 

   4

   
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/producer1.keytab" \
       principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   1

   Kerberos (GSSAPI) 認証の設定。

   2

   Kerberos は SASL プレーンテキスト (ユーザー名/パスワード) セキュリティープロトコルを使用します。

   3

   Kerberos KDC で設定された Kafka のサービスプリンシパル (ユーザー)。

   4

   jaas.conf で定義されたものと同じプロパティーを使用した JAAS の設定。

   /opt/kafka/config/consumer.properties

   Copy to Clipboard Toggle word wrap

   # ...
   sasl.mechanism=GSSAPI
   security.protocol=SASL_PLAINTEXT
   sasl.kerberos.service.name=kafka
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/consumer1.keytab" \
       principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

2. クライアントを実行して、Kafka ブローカーからメッセージを送受信できることを確認します。

   プロデューサークライアント:

   Copy to Clipboard Toggle word wrap

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   コンシューマークライアント:

   Copy to Clipboard Toggle word wrap

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

SASL DIGEST-MD5 認証の有効化

1. すべての Kafka ブローカーノードで、/opt/kafka/config/jaas.conf JAAS 設定ファイルを作成または編集し、以下のコンテキストを追加します。

   Copy to Clipboard Toggle word wrap

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="<Username>"
       password="<Password>";
   };

   ユーザー名とパスワードは ZooKeeper で設定されているものと同じである必要があります。

   Client コンテキストの例を以下に示します。

   Copy to Clipboard Toggle word wrap

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

2. すべての Kafka ブローカーノードを 1 つずつ再起動します。JAAS 設定を Kafka ブローカーに渡すには、KAFKA_OPTS 環境変数を使用します。

   Copy to Clipboard Toggle word wrap

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

   マルチノードクラスターでブローカーを再起動する方法は、「Kafka ブローカーの正常なローリング再起動の実行」 を参照してください。

手順

1. Kafka 設定プロパティーファイルを編集して、すべてのクラスターノードで zookeeper.set.acl フィールドを true に設定します。

   Copy to Clipboard Toggle word wrap

   zookeeper.set.acl=true

2. Kafka ブローカーを起動します。

ZooKeeper ACL の有効化

1. Kafka 設定プロパティーファイルを編集して、すべてのクラスターノードで zookeeper.set.acl フィールドを true に設定します。

   Copy to Clipboard Toggle word wrap

   zookeeper.set.acl=true

2. すべての Kafka ブローカーを 1 つずつ再起動します。

   マルチノードクラスターでブローカーを再起動する方法は、「Kafka ブローカーの正常なローリング再起動の実行」 を参照してください。

3. zookeeper-security-migration.sh ツールを使用して、既存のすべての ZooKeeper znodes に ACL を設定します。

   Copy to Clipboard Toggle word wrap

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=<ZooKeeperURL>
   exit

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=zoo1.my-domain.com:2181
   exit

1. Streams for Apache Kafka クラスターおよび Kafka クライアントの OAuth 2.0 認可サーバーを設定します。
2. OAuth 2.0 を使用するよう設定された Kafka ブローカーリスナーで Kafka クラスターをデプロイまたは更新します。
3. OAuth 2.0 を使用するように Java ベースの Kafka クライアントを更新します。

手順

1. Red Hat Single Sign-On の Admin Console にアクセスするか、Red Hat Single Sign-On の Admin CLI を使用して、OAuth 2.0 認証の設定時に作成した Kafka ブローカークライアントの Authorization Services を有効にします。
2. 認可サービスを使用して、クライアントのリソース、認可スコープ、ポリシー、およびパーミッションを定義します。
3. ロールとグループをユーザーとクライアントに割り当てて、パーミッションをユーザーとクライアントにバインドします。

4. Red Hat Single Sign-On 認可を使用するように Kafka ブローカーを設定します。

   以下を Kafka server.properties 設定ファイルに追加し、Kafka に authorizer をインストールします。

   Copy to Clipboard Toggle word wrap

   authorizer.class.name=io.strimzi.kafka.oauth.server.authorizer.KeycloakAuthorizer
   principal.builder.class=io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder

5. Kafka ブローカーの設定を追加して、認可サーバーおよび Authorization Services にアクセスします。

   ここでは、server.properties への追加プロパティーとして追加される設定例を示しますが、大文字で始める、または大文字の命名規則を使用して、環境変数として定義することもできます。

   Copy to Clipboard Toggle word wrap

   strimzi.authorization.token.endpoint.uri="https://<auth_server_address>/auth/realms/REALM-NAME/protocol/openid-connect/token" 

   1

   
   strimzi.authorization.client.id="kafka" 

   2

   1

   Red Hat Single Sign-On への OAuth 2.0 トークンエンドポイントの URL。実稼働環境の場合は、常に https:// urls を使用してください。

   2

   Authorization Services が有効な Red Hat Single Sign-On の OAuth 2.0 クライアント定義のクライアント ID。通常、kafka が ID として使用されます。

6. (オプション) 特定の Kafka クラスターの設定を追加します。

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   strimzi.authorization.kafka.cluster.name="kafka-cluster" 

   1

   1

   特定の Kafka クラスターの名前。名前はパーミッションをターゲットにするために使用され、同じ Red Hat シングルサインオンレルム内で複数のクラスターを管理できるようにします。デフォルト値は kafka-cluster です。

7. (オプション) シンプルな認可に委譲します。

   Copy to Clipboard Toggle word wrap

   strimzi.authorization.delegate.to.kafka.acl="true" 

   1

   1

   Red Hat Single Sign-On の Authorization Services ポリシーによってアクセスが拒否された場合に、Kafka AclAuthorizer に認可を委譲します。デフォルトは false です。

8. (オプション) TLS 接続の設定を認可サーバーに追加します。

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   strimzi.authorization.ssl.truststore.location=<path_to_truststore> 

   1

   
   strimzi.authorization.ssl.truststore.password=<my_truststore_password> 

   2

   
   strimzi.authorization.ssl.truststore.type=JKS 

   3

   
   strimzi.authorization.ssl.secure.random.implementation=SHA1PRNG 

   4

   
   strimzi.authorization.ssl.endpoint.identification.algorithm=HTTPS 

   5

   1

   証明書が含まれるトラストストアへのパス。

   2

   トラストストアのパスワード。

   3

   トラストストアのタイプ。設定されていない場合は、デフォルトの Java キーストアタイプが使用されます。

   4

   乱数ジェネレーターの実装。設定されていない場合は、Java プラットフォーム SDK デフォルトが使用されます。

   5

   ホスト名の検証。空の文字列に設定すると、ホスト名の検証はオフになります。設定されていない場合、デフォルト値は HTTPS で、サーバー証明書のホスト名の検証を強制します。

9. (オプション) 認可サーバーからの付与の更新を設定します。付与更新ジョブは、アクティブなトークンを列挙し、それぞれに最新の付与を要求することで機能します。

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   strimzi.authorization.grants.refresh.period.seconds="120" 

   1

   
   strimzi.authorization.grants.refresh.pool.size="10" 

   2

   
   strimzi.authorization.grants.max.idle.time.seconds="300" 

   3

   
   strimzi.authorization.grants.gc.period.seconds="300" 

   4

   
   strimzi.authorization.reuse.grants="false" 

   5

   1

   認可サーバーからの付与のリストが更新される頻度を指定します (デフォルトでは 1 分に 1 回)。デバッグの目的で付与の更新をオフにするには、"0" に設定します。

   2

   付与更新ジョブで使用されるスレッドプールのサイズ (並列度) を指定します。デフォルト値は "5" です。

   3

   キャッシュ内のアイドル許可を削除できるようになるまでの時間 (秒単位)。デフォルト値は 300 です。

   4

   キャッシュから古い許可を削除するジョブの連続実行間の時間 (秒単位)。デフォルト値は 300 です。

   5

   新しいセッションに対して最新の許可を取得するかどうかを制御します。無効にすると、許可が Red Hat Single Sign-On から取得され、ユーザーのためにキャッシュされます。デフォルト値は true です。

10. (オプション) 認可サーバーとの通信時のネットワークタイムアウトを設定します。

    以下に例を示します。

    Copy to Clipboard Toggle word wrap

    strimzi.authorization.connect.timeout.seconds="60" 

    1

    
    strimzi.authorization.read.timeout.seconds="60" 

    2

    
    strimzi.authorization.http.retries="2" 

    3

    1

    Red Hat Single Sign-On トークンエンドポイントに接続するときの接続タイムアウト (秒単位)。デフォルト値は 60 です。

    2

    Red Hat Single Sign-On トークンエンドポイントに接続するときの読み取りタイムアウト (秒単位)。デフォルト値は 60 です。

    3

    認可サーバーへの失敗した HTTP リクエストを (一時停止せずに) 再試行する最大回数。デフォルト値は 0 で、再試行は実行されないことを意味します。このオプションを効果的に使用するには、strimzi.authorization.connect.timeout.seconds オプションと strimzi.authorization.read.timeout.seconds オプションのタイムアウト時間を短縮することを検討してください。ただし、再試行により現在のワーカースレッドが他のリクエストで利用できなくなる可能性があり、リクエストが多すぎると停止する場合、Kafka ブローカーが応答しなくなる可能性があることに注意してください。

11. (オプション) トークンの検証と認可のために OAuth 2.0 メトリクスを有効にします。

    Copy to Clipboard Toggle word wrap

    oauth.enable.metrics="true" 

    1

    1

    OAuth メトリクスを有効にするか無効にするかを制御します。デフォルト値は false です。

12. (オプション) リクエストから Accept ヘッダーを削除します。

    Copy to Clipboard Toggle word wrap

    oauth.include.accept.header="false" 

    1

    1

    ヘッダーを含めることで認可サーバーとの通信時に問題が発生する場合は、false に設定します。デフォルト値は true です。

13. クライアントまたは特定のロールを持つユーザーとして Kafka ブローカーにアクセスして、設定したパーミッションを検証し、必要なアクセス権限があり、付与されるべきでないアクセス権限がないことを確認します。

手順

1. Kafka ブローカーで操作を実行するためのクライアントリクエストを承認するのに必要な OPA ポリシーを記述します。

   OPA ポリシーの定義 を参照してください。

   これで、Kafka ブローカーが OPA を使用するように設定します。

2. Kafka の OPA authorizer プラグイン をインストールします。

   OPA への接続 を参照してください。

   プラグインファイルが Kafka クラスパスに含まれていることを確認してください。

3. 以下を Kafka server.properties 設定ファイルに追加し、OPA プラグインを有効にします。

   Copy to Clipboard Toggle word wrap

   authorizer.class.name: com.bisnode.kafka.authorization.OpaAuthorizer

4. Kafka ブローカーの server.properties に設定をさらに追加して、OPA ポリシーエンジンおよびポリシーにアクセスします。

   以下に例を示します。

   Copy to Clipboard Toggle word wrap

   opa.authorizer.url=https://OPA-ADDRESS/allow 

   1

   
   opa.authorizer.allow.on.error=false 

   2

   
   opa.authorizer.cache.initial.capacity=50000 

   3

   
   opa.authorizer.cache.maximum.size=50000 

   4

   
   opa.authorizer.cache.expire.after.seconds=600000 

   5

   
   super.users=User:alice;User:bob 

   6

   1

   (必須) authorizer プラグインがクエリーするポリシーの OAuth 2.0 トークンエンドポイント URL。この例では、ポリシーは allow という名前です。

   2

   authorizer プラグインが OPA ポリシーエンジンとの接続に失敗した場合に、クライアントがデフォルトで許可または拒否されるかどうかを指定するフラグ。

   3

   ローカルキャッシュの初期容量 (バイト単位)。すべてのリクエストについてプラグインに OPA ポリシーエンジンをクエリーする必要がないように、キャッシュが使用されます。

   4

   ローカルキャッシュの最大容量 (バイト単位)。

   5

   OPA ポリシーエンジンからのリロードによってローカルキャッシュが更新される時間 (ミリ秒単位)。

   6

   スーパーユーザーとして扱われるユーザープリンシパルのリスト。これにより、Open Policy Agent ポリシーをクエリーしなくても常に許可されます。

   認証および認可オプションの詳細は、Open Policy Agent の Web サイト を参照してください。

5. 適切な認可を持つクライアントと持たないクライアントを使用して Kafka ブローカーにアクセスし、設定した権限を確認します。

手順

1. /opt/kafka/config/connect-standalone.properties Kafka Connect 設定ファイルを編集し、bootstrap.server が Kafka ブローカーを指すように設定します。以下に例を示します。

   Copy to Clipboard Toggle word wrap

   bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092

2. 設定ファイルで Kafka Connect を起動し、1 つ以上のコネクター設定を指定します。

   Copy to Clipboard Toggle word wrap

   su - kafka
   /opt/kafka/bin/connect-standalone.sh /opt/kafka/config/connect-standalone.properties connector1.properties
   [connector2.properties ...]

3. Kafka Connect が実行されていることを確認します。

   Copy to Clipboard Toggle word wrap

   jcmd | grep ConnectStandalone

クラスターの実行

1. すべての Kafka Connect ワーカーノードで /opt/kafka/config/connect-distributed.properties Kafka Connect 設定ファイルを編集します。

   • bootstrap.server オプションを設定して、Kafka ブローカーを示すようにします。
   • group.id オプションを設定します。
   • config.storage.topic オプションを設定します。
   • offset.storage.topic オプションを設定します。

   • status.storage.topic オプションを設定します。

     以下に例を示します。

     Copy to Clipboard Toggle word wrap

     bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092
     group.id=my-group-id
     config.storage.topic=my-group-id-configs
     offset.storage.topic=my-group-id-offsets
     status.storage.topic=my-group-id-status

2. すべての Kafka Connect ワーカーノードで /opt/kafka/config/connect-distributed.properties Kafka Connect ワーカーを起動します。

   Copy to Clipboard Toggle word wrap

   su - kafka
   /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

3. Kafka Connect が実行されていることを確認します。

   Copy to Clipboard Toggle word wrap

   jcmd | grep ConnectDistributed

4. Kafka Connect REST API を使用して コネクターを管理 します。

手順

1. コネクター設定で JSON ペイロードを準備します。以下に例を示します。

   Copy to Clipboard Toggle word wrap

   {
     "name": "my-connector",
     "config": {
     "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
       "tasks.max": "1",
       "topics": "my-topic-1,my-topic-2",
       "file": "/tmp/output-file.txt"
     }
   }

2. POST リクエストを <KafkaConnectAddress>:8083/connectors に送信してコネクターを作成します。以下の例では、curl を使用します。

   Copy to Clipboard Toggle word wrap

   curl -X POST -H "Content-Type: application/json" --data @sink-connector.json http://connect0.my-domain.com:8083/connectors

3. <KafkaConnectAddress>:8083/connectors に GET リクエストを送信して、コネクターがデプロイされたことを確認します。以下の例では、curl を使用します。

   Copy to Clipboard Toggle word wrap

   curl http://connect0.my-domain.com:8083/connectors

コネクターの削除

1. <KafkaConnectAddress>:8083/connectors/<ConnectorName> に GET リクエストを送信して、コネクターが存在することを確認します。以下の例では、curl を使用します。

   Copy to Clipboard Toggle word wrap

   curl http://connect0.my-domain.com:8083/connectors

2. コネクターを削除するには、DELETE リクエストを <KafkaConnectAddress>:8083/connectors に送信します。以下の例では、curl を使用します。

   Copy to Clipboard Toggle word wrap

   curl -X DELETE http://connect0.my-domain.com:8083/connectors/my-connector

3. <KafkaConnectAddress>:8083/connectors に GET リクエストを送信して、コネクターが削除されたことを確認します。以下の例では、curl を使用します。

   Copy to Clipboard Toggle word wrap

   curl http://connect0.my-domain.com:8083/connectors