15.4. OAuth 2.0 トークンベース認証の使用

1. Kafka クライアントは、クライアント ID とシークレット、および必要に応じてリフレッシュトークンを使用して、認可サーバーからアクセストークンを要求します。または、クライアントはユーザー名とパスワードを使用して認証することもできます。
2. 認可サーバーは新しいアクセストークンを生成します。
3. Kafka クライアントは、SASL OAUTHBEARER メカニズムを使用してアクセストークンを渡すことで Kafka ブローカーで認証されます。
4. Kafka ブローカーは、独自のクライアント ID およびシークレットを使用し、認可サーバーでトークンイントロスペクションエンドポイントを呼び出すことで、アクセストークンを検証します。
5. トークンが有効な場合、Kafka クライアントセッションが確立されます。

1. Kafka クライアントは、クライアント ID とシークレット、および必要に応じてリフレッシュトークンを使用して、トークンエンドポイントから認可サーバーに対して認証を行います。または、クライアントはユーザー名とパスワードを使用して認証することもできます。
2. 認可サーバーは新しいアクセストークンを生成します。
3. Kafka クライアントは、SASL OAUTHBEARER メカニズムを使用してアクセストークンを渡すことで Kafka ブローカーで認証されます。
4. Kafka ブローカーは、JWT トークン署名チェックおよびローカルトークンイントロスペクションを使用して、ローカルでアクセストークンを検証します。

1. Kafka クライアントは、SASL OAUTHBEARER メカニズムを使用して、有効期限の長いアクセストークンを渡すために Kafka ブローカーで認証します。
2. Kafka ブローカーは、独自のクライアント ID およびシークレットを使用して、認可サーバーでトークンイントロスペクションエンドポイントを呼び出し、アクセストークンを検証します。
3. トークンが有効な場合、Kafka クライアントセッションが確立されます。

1. Kafka クライアントは、SASL OAUTHBEARER メカニズムを使用して、有効期限の長いアクセストークンを渡すために Kafka ブローカーで認証します。
2. Kafka ブローカーは、JWT トークン署名チェックおよびローカルトークンイントロスペクションを使用して、ローカルでアクセストークンを検証します。

1. Kafka クライアントは、clientId をユーザー名として、secret をパスワードとして渡します。
2. Kafka ブローカーは、トークンエンドポイントを使用して clientId および secret を認可サーバーに渡します。
3. 認可サーバーは、新しいアクセストークンまたはエラー (クライアントクレデンシャルが有効でない場合) を返します。

4. Kafka ブローカーは、以下のいずれかの方法でトークンを検証します。

   1. トークンイントロスペクションエンドポイントが指定されている場合、Kafka ブローカーは認可サーバーでエンドポイントを呼び出すことで、アクセストークンを検証します。トークンの検証に成功した場合には、セッションが確立されます。
   2. ローカルトークンのイントロスペクションが使用される場合、要求は認可サーバーに対して行われません。Kafka ブローカーは、JWT トークン署名チェックを使用して、アクセストークンをローカルで検証します。

1. Kafka クライアントはユーザー名とパスワードを渡します。パスワードは、クライアントを実行する前に手動で取得および設定されたアクセストークンの値を提供します。

2. Kafka ブローカーリスナーが認証のトークンエンドポイントで設定されているかどうかに応じて、$accessToken: 文字列の接頭辞の有無にかかわらず、パスワードは渡されます。

   1. トークンエンドポイントが設定されている場合、パスワードの前に $accessToken: を付け、password パラメーターにクライアントシークレットではなくアクセストークンが含まれていることをブローカーに知らせる必要があります。Kafka ブローカーは、ユーザー名をアカウントのユーザー名として解釈します。
   2. トークンエンドポイントが Kafka ブローカーリスナーで設定されていない場合 (no-client-credentials mode を強制)、パスワードは接頭辞なしでアクセストークンを提供する必要があります。Kafka ブローカーは、ユーザー名をアカウントのユーザー名として解釈します。このモードでは、クライアントはクライアント ID およびシークレットを使用せず、password パラメーターは常に raw アクセストークンとして解釈されます。

3. Kafka ブローカーは、以下のいずれかの方法でトークンを検証します。

   1. トークンイントロスペクションエンドポイントが指定されている場合、Kafka ブローカーは認可サーバーでエンドポイントを呼び出すことで、アクセストークンを検証します。トークンの検証に成功した場合には、セッションが確立されます。
   2. ローカルトークンイントロスペクションが使用されている場合には、認可サーバーへの要求はありません。Kafka ブローカーは、JWT トークン署名チェックを使用して、アクセストークンをローカルで検証します。

手順

1. 認可サーバーをクラスターにデプロイします。

2. 認可サーバーの CLI または管理コンソールにアクセスして、Streams for Apache Kafka 用に OAuth 2.0 を設定します。

   Streams for Apache Kafka で動作するように認可サーバーを準備します。

3. kafka-broker クライアントを設定します。
4. アプリケーションの Kafka クライアントコンポーネントごとにクライアントを設定します。

手順

1. エディターで、Kafka リソースの Kafka ブローカー設定 (Kafka.spec.kafka) を更新します。

   oc edit kafka my-cluster

2. Kafka ブローカーの listeners 設定を行います。

   各タイプのリスナーは独立しているため、同じ設定にする必要はありません。

   以下は、外部リスナーに設定された設定オプションの例になります。

   例 1: 高速なローカル JWT トークン検証の設定

   #...
   - name: external3
     port: 9094
     type: loadbalancer
     tls: true
     authentication:
       type: oauth 1
       validIssuerUri: https://<auth_server_address>/auth/realms/external 2
       jwksEndpointUri: https://<auth_server_address>/auth/realms/external/protocol/openid-connect/certs 3
       userNameClaim: preferred_username 4
       maxSecondsWithoutReauthentication: 3600 5
       tlsTrustedCertificates: 6
       - secretName: oauth-server-cert
         certificate: ca.crt
       disableTlsHostnameVerification: true 7
       jwksExpirySeconds: 360 8
       jwksRefreshSeconds: 300 9
       jwksMinRefreshPauseSeconds: 1 10

   1

   oauth に設定されたリスナータイプ。

   2

   認証に使用されるトークン発行者の URI。

   3

   ローカルの JWT 検証に使用される JWKS 証明書エンドポイントの URI。

   4

   ユーザーを識別するために使用される実際のユーザー名を含むトークンクレーム (またはキー)。その値は、認可サーバーによって異なります。必要に応じて、"['user.info'].['user.id']" のような JsonPath 式を使用して、トークン内のネストされた JSON 属性からユーザー名を取得できます。

   5

   (任意設定): セッションの有効期限がアクセストークンと同じ期間になるよう強制する Kafka の再認証メカニズムを有効にします。指定された値がアクセストークンの有効期限が切れるまでの残り時間よりも短い場合、クライアントは実際にトークンの有効期限が切れる前に再認証する必要があります。デフォルトでは、アクセストークンの期限が切れてもセッションは期限切れにならず、クライアントは再認証を試行しません。

   6

   (任意設定): 認可サーバーへの TLS 接続用の信用できる証明書。

   7

   (任意設定): TLS ホスト名の検証を無効にします。デフォルトは false です。

   8

   JWKS 証明書が期限切れになる前に有効であるとみなされる期間。デフォルトは 360 秒です。デフォルトよりも長い時間を指定する場合は、無効になった証明書へのアクセスが許可されるリスクを考慮してください。

   9

   JWKS 証明書を更新する間隔。この間隔は、有効期間よりも 60 秒以上短くする必要があります。デフォルトは 300 秒です。

   10

   JWKS 公開鍵の更新が連続して試行される間隔の最小一時停止時間 (秒単位)。不明な署名キーが検出されると、JWKS キーの更新は、最後に更新を試みてから少なくとも指定された期間は一時停止し、通常の定期スケジュール以外でスケジュールされます。キーの更新は指数バックオフ (指数バックオフ) のルールに従い、jwksRefreshSeconds に到達するまで、一時停止を増やして失敗した更新を再試行します。デフォルト値は 1 です。

   例 2: イントロスペクションエンドポイントを使用したトークンの検証の設定

   - name: external3
     port: 9094
     type: loadbalancer
     tls: true
     authentication:
       type: oauth
       validIssuerUri: https://<auth_server_address>/auth/realms/external
       introspectionEndpointUri: https://<auth_server_address>/auth/realms/external/protocol/openid-connect/token/introspect 1
       clientId: kafka-broker 2
       clientSecret: 3
         secretName: my-cluster-oauth
         key: clientSecret
       userNameClaim: preferred_username 4
       maxSecondsWithoutReauthentication: 3600 5

   1

   トークンイントロスペクションエンドポイントの URI。

   2

   クライアントを識別するためのクライアント ID。

   3

   認証にはクライアントシークレットとクライアント ID が使用されます。

   4

   ユーザーを識別するために使用される実際のユーザー名を含むトークンクレーム (またはキー)。その値は、認可サーバーによって異なります。必要に応じて、"['user.info'].['user.id']" のような JsonPath 式を使用して、トークン内のネストされた JSON 属性からユーザー名を取得できます。

   5

   (任意設定): セッションの有効期限がアクセストークンと同じ期間になるよう強制する Kafka の再認証メカニズムを有効にします。指定された値がアクセストークンの有効期限が切れるまでの残り時間よりも短い場合、クライアントは実際にトークンの有効期限が切れる前に再認証する必要があります。デフォルトでは、アクセストークンの期限が切れてもセッションは期限切れにならず、クライアントは再認証を試行しません。

   OAuth 2.0 認証の適用方法や、認可サーバーのタイプによっては、追加 (任意) の設定を使用できます。

     # ...
     authentication:
       type: oauth
       # ...
       checkIssuer: false 1
       checkAudience: true 2
       fallbackUserNameClaim: client_id 3
       fallbackUserNamePrefix: client-account- 4
       validTokenType: bearer 5
       userInfoEndpointUri: https://<auth_server_address>/auth/realms/external/protocol/openid-connect/userinfo 6
       enableOauthBearer: false 7
       enablePlain: true 8
       tokenEndpointUri: https://<auth_server_address>/auth/realms/external/protocol/openid-connect/token 9
       customClaimCheck: "@.custom == 'custom-value'" 10
       clientAudience: audience 11
       clientScope: scope 12
       connectTimeoutSeconds: 60 13
       readTimeoutSeconds: 60 14
       httpRetries: 2 15
       httpRetryPauseMs: 300 16
       groupsClaim: "$.groups" 17
       groupsClaimDelimiter: "," 18
       includeAcceptHeader: false 19

   1

   認可サーバーが iss クレームを提供しない場合は、発行者チェックを行うことができません。このような場合、checkIssuer を false に設定し、validIssuerUri を指定しないようにします。デフォルトは true です。

   2

   オーソリゼーションサーバーが aud(オーディエンス) クレームを提供していて、オーディエンスチェックを実施したい場合は、checkAudience を true に設定します。オーディエンスチェックによって、トークンの目的の受信者が特定されます。これにより、Kafka ブローカーは aud 要求に clientId を持たないトークンを拒否します。デフォルトは false です。

   3

   認可サーバーは、通常ユーザーとクライアントの両方を識別する単一の属性を提供しない場合があります。クライアントが独自の名前で認証される場合、サーバーによって クライアント ID が提供されることがあります。更新トークンまたはアクセストークンを取得するために、ユーザー名およびパスワードを使用してユーザーが認証される場合、サーバーによってクライアント ID の他に username が提供されることがあります。プライマリーユーザー ID 属性が使用できない場合は、このフォールバックオプションで、使用するユーザー名クレーム (属性) を指定します。必要に応じて、"['client.info'].['client.id']" のような JsonPath 式を使用してフォールバックユーザー名を取得し、トークン内のネストされた JSON 属性からユーザー名を取得できます。

   4

   fallbackUserNameClaim が適用される場合、ユーザー名クレームの値とフォールバックユーザー名クレームの値が競合しないようにする必要もあることがあります。producer というクライアントが存在し、producer という通常ユーザーも存在する場合について考えてみましょう。この 2 つを区別するには、このプロパティーを使用してクライアントのユーザー ID に接頭辞を追加します。

   5

   (introspectionEndpointUri を使用する場合のみ該当): 使用している認可サーバーによっては、イントロスペクションエンドポイントによってトークンタイプ属性が返されるかどうかは分からず、異なる値が含まれることがあります。イントロスペクションエンドポイントからの応答に含まれなければならない有効なトークンタイプ値を指定できます。

   6

   (introspectionEndpointUri を使用する場合のみ該当): イントロスペクションエンドポイントの応答に識別可能な情報が含まれないように、認可サーバーが設定または実装されることがあります。ユーザー ID を取得するには、userinfo エンドポイントの URI をフォールバックとして設定します。userNameClaim、fallbackUserNameClaim、および fallbackUserNamePrefix の設定が userinfo エンドポイントの応答に適用されます。

   7

   これを false に設定してリスナーで OAUTHBEARER メカニズムを無効にします。PLAIN または OAUTHBEARER のいずれかを有効にする必要があります。デフォルトは true です。

   8

   リスナーで PLAIN 認証を有効にするには、true に設定します。これは、すべてのプラットフォームのすべてのクライアントでサポートされています。

   9

   PLAIN メカニズムの追加設定。これが指定されている場合、クライアントは $accessToken: 接頭辞を使用してアクセストークンを password として渡すことで、PLAIN 経由で認証できます。実稼働環境の場合は、常に https:// urls を使用してください。

   10

   これを JsonPath フィルタークエリーに設定すると、検証中に追加のカスタムルールを JWT アクセストークンに適用できます。アクセストークンに必要なデータが含まれていないと拒否されます。introspectionEndpointUri を使用する場合、カスタムチェックはイントロスペクションエンドポイントの応答 JSON に適用されます。

   11

   トークンエンドポイントに渡される audience パラメーター。オーディエンス は、inter-broker 認証用にアクセストークンを取得する場合に使用されます。また、clientId と secret を使用した PLAIN クライアント認証の上にある OAuth 2.0 のクライアント名にも使われています。これは、認可サーバーに応じて、トークンの取得機能とトークンの内容のみに影響します。リスナーによるトークン検証ルールには影響しません。

   12

   scope パラメーターがトークンエンドポイントに渡されます。スコープ は、inter-broker 認証用にアクセストークンを取得する場合に使用されます。また、clientId と secret を使用した PLAIN クライアント認証の上にある OAuth 2.0 のクライアント名にも使われています。これは、認可サーバーに応じて、トークンの取得機能とトークンの内容のみに影響します。リスナーによるトークン検証ルールには影響しません。

   13

   認可サーバーへの接続時のタイムアウト (秒単位)。デフォルト値は 60 です。

   14

   認可サーバーへの接続時の読み取りタイムアウト (秒単位)。デフォルト値は 60 です。

   15

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

   16

   認可サーバーへの失敗した HTTP リクエストの再試行を行うまでの待機時間。デフォルトでは、この時間はゼロに設定されており、一時停止は適用されません。これは、リクエストの失敗の原因となる問題の多くは、リクエストごとのネットワークの不具合やプロキシーの問題であり、すぐに解決できるためです。ただし、認可サーバーに負荷がかかっている場合、または高トラフィックが発生している場合は、このオプションを 100 ミリ秒以上の値に設定して、サーバーの負荷を軽減し、再試行が成功する可能性を高めることができます。

   17

   JWT トークンまたはイントロスペクションエンドポイント応答からグループ情報を抽出するために使用される JsonPath クエリー。このオプションはデフォルトでは設定されていません。このオプションを設定すると、カスタム承認者はユーザーグループに基づいて承認の決定を行うことができます。

   18

   グループ情報が単一の区切り文字列として返される場合に、グループ情報を解析するために使用される区切り文字。デフォルト値は ',' (コンマ) です。

   19

   一部の認可サーバーでは、クライアントが Accept: application/json ヘッダーを送信する際に問題が発生します。includeAcceptHeader: false を設定する場合、ヘッダーは送信されません。デフォルトは true です。

3. エディターを保存して終了し、ローリング更新の完了を待ちます。

4. 更新をログで確認するか、Pod 状態の遷移を監視して確認します。

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   ローリング更新によって、ブローカーが OAuth 2.0 認証を使用するように設定されます。

手順

1. OAuth 2.0 サポートのあるクライアントライブラリーを Kafka クライアントの pom.xml ファイルに追加します。

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.15.0.redhat-00007</version>
   </dependency>

2. プロパティーファイルで以下の設定を指定して、クライアントプロパティーを設定します。

   • セキュリティープロトコル
   • SASL メカニズム

   • 使用されているメソッドに応じた JAAS モジュールと認証プロパティー

     たとえば、以下を client.properties ファイルに追加できます。

     クライアントクレデンシャルメカニズムのプロパティー

     security.protocol=SASL_SSL 1
     sasl.mechanism=OAUTHBEARER 2
     ssl.truststore.location=/tmp/truststore.p12 3
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \ 4
       oauth.client.id="<client_id>" \ 5
       oauth.client.secret="<client_secret>" \ 6
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 7
       oauth.ssl.truststore.password="$STOREPASS" \ 8
       oauth.ssl.truststore.type="PKCS12" \ 9
       oauth.scope="<scope>" \ 10
       oauth.audience="<audience>" ; 11
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

     TLS 暗号化接続用の SASL_SSL セキュリティープロトコル。ローカル開発のみでは、暗号化されていない接続で SASL_PLAINTEXT を使用します。

     2

     OAUTHBEARER または PLAIN として指定された SASL メカニズム。

     3

     Kafka クラスターへのセキュアなアクセスのためのトラストストア設定。

     4

     認可サーバーのトークンエンドポイントの URI です。

     5

     クライアント ID。認可サーバーで クライアント を作成するときに使用される名前です。

     6

     認可サーバーで クライアント を作成するときに作成されるクライアントシークレット。

     7

     この場所には、認可サーバーの公開鍵証明書 (truststore.p12) が含まれています。

     8

     トラストストアにアクセスするためのパスワード。

     9

     トラストストアのタイプ。

     10

     (オプション): トークンエンドポイントからトークンを要求するための scope。認可サーバーでは、クライアントによるスコープの指定が必要になることがあります。

     11

     (オプション) トークンエンドポイントからトークンを要求するための audience。認可サーバーでは、クライアントによるオーディエンスの指定が必要になることがあります。

     パスワード付与メカニズムのプロパティー

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 1
       oauth.client.secret="<client_secret>" \ 2
       oauth.password.grant.username="<username>" \ 3
       oauth.password.grant.password="<password>" \ 4
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \
       oauth.scope="<scope>" \
       oauth.audience="<audience>" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

     クライアント ID。認可サーバーで クライアント を作成するときに使用される名前です。

     2

     (オプション) 認可サーバーで クライアント を作成するときに作成されるクライアントシークレット。

     3

     パスワードグラント認証のユーザー名。OAuth パスワードグラント設定 (ユーザー名とパスワード) では、OAuth 2.0 パスワードグラント方式が使用されます。パスワードグラントを使用するには、権限が制限されたクライアントのユーザーアカウントを認可サーバー上に作成します。アカウントは、サービスアカウントのように機能する必要があります。認証にユーザーアカウントが必要な環境で使用しますが、先にリフレッシュトークンの使用を検討してください。

     4

     パスワード付与認証のパスワード。

     注記

     SASL PLAIN は、OAuth 2.0 パスワード付与メソッドを使用したユーザー名とパスワードの受け渡し (パスワード付与) をサポートしていません。

     アクセストークンのプロパティー

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.access.token="<access_token>" \ 1
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

     Kafka クライアントの有効期間が長いアクセストークン。

     トークンのプロパティーを更新する

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 1
       oauth.client.secret="<client_secret>" \ 2
       oauth.refresh.token="<refresh_token>" \ 3
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

     クライアント ID。認可サーバーで クライアント を作成するときに使用される名前です。

     2

     (オプション) 認可サーバーで クライアント を作成するときに作成されるクライアントシークレット。

     3

     Kafka クライアントの有効期間が長い更新トークン。

3. OAUTH 2.0 認証のクライアントプロパティーを Java クライアントコードに入力します。

   クライアントプロパティーの入力を示す例

   Properties props = new Properties();
   try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
     props.load(reader);
   }

4. Kafka クライアントが Kafka ブローカーにアクセスできることを確認します。

手順

1. クライアントシークレットを作成し、これを環境変数としてコンポーネントにマウントします。

   以下は、Kafka Bridge の Secret を作成する例になります。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Secret
   metadata:
    name: my-bridge-oauth
   type: Opaque
   data:
    clientSecret: MGQ1OTRmMzYtZTllZS00MDY2LWI5OGEtMTM5MzM2NjdlZjQw 1

   1

   clientSecret キーは base64 形式である必要があります。

2. Kafka コンポーネントのリソースを作成または編集し、OAuth 2.0 認証が認証プロパティーに設定されるようにします。

   OAuth 2.0 認証の場合、次のオプションを使用できます。

   • クライアント ID およびシークレット
   • クライアント ID およびリフレッシュトークン
   • アクセストークン
   • ユーザー名およびパスワード
   • TLS

   以下は、クライアント ID、シークレット、および TLS を使用して OAuth 2.0 が Kafka Bridge クライアントに割り当てられる例になります。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     # ...
     authentication:
       type: oauth 1
       tokenEndpointUri: https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token 2
       clientId: kafka-bridge
       clientSecret:
         secretName: my-bridge-oauth
         key: clientSecret
       tlsTrustedCertificates: 3
       - secretName: oauth-server-cert
         certificate: tls.crt

   1

   oauth に設定された認証タイプ。

   2

   認証用のトークンエンドポイントの URI。

   3

   認可サーバーへの TLS 接続用の信用できる証明書。

   OAuth 2.0 認証の適用方法や、認可サーバーのタイプによって、使用できる追加の設定オプションがあります。

   # ...
   spec:
     # ...
     authentication:
       # ...
       disableTlsHostnameVerification: true 1
       checkAccessTokenType: false 2
       accessTokenIsJwt: false 3
       scope: any 4
       audience: kafka 5
       connectTimeoutSeconds: 60 6
       readTimeoutSeconds: 60 7
       httpRetries: 2 8
       httpRetryPauseMs: 300 9
       includeAcceptHeader: false 10

   1

   (任意設定): TLS ホスト名の検証を無効にします。デフォルトは false です。

   2

   認可サーバーによって、JWT トークン内部で typ (タイプ) 要求が返されない場合は、checkAccessTokenType: false を適用するとトークンタイプがチェックされず次に進むことができます。デフォルトは true です。

   3

   不透明なトークンを使用している場合、アクセストークンが JWT トークンとして処理されないように accessTokenIsJwt: false を適用することができます。

   4

   (オプション): トークンエンドポイントからトークンを要求するための scope。認可サーバーでは、クライアントによるスコープの指定が必要になることがあります。この場合では any になります。

   5

   (オプション) トークンエンドポイントからトークンを要求するための audience。認可サーバーでは、クライアントによるオーディエンスの指定が必要になることがあります。今回の場合は kafka です。

   6

   (オプション) 認可サーバーへの接続時のタイムアウト (秒単位)。デフォルト値は 60 です。

   7

   (オプション): 認可サーバーへの接続時の読み取りタイムアウト (秒単位)。デフォルト値は 60 です。

   8

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

   9

   (オプション) 失敗した認可サーバーへの HTTP リクエストの再試行を行うまでの待機時間。デフォルトでは、この時間はゼロに設定されており、一時停止は適用されません。これは、リクエストの失敗の原因となる問題の多くは、リクエストごとのネットワークの不具合やプロキシーの問題であり、すぐに解決できるためです。ただし、認可サーバーに負荷がかかっている場合、または高トラフィックが発生している場合は、このオプションを 100 ミリ秒以上の値に設定して、サーバーの負荷を軽減し、再試行が成功する可能性を高めることができます。

   10

   (オプション) 一部の認可サーバーでは、クライアントが Accept: application/json ヘッダーを送信する際に問題が発生します。includeAcceptHeader: false を設定する場合、ヘッダーは送信されません。デフォルトは true です。

3. Kafka リソースのデプロイメントに変更を適用します。

   oc apply -f your-file

4. 更新をログで確認するか、Pod 状態の遷移を監視して確認します。

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   ローリング更新では、OAuth 2.0 認証を使用して Kafka ブローカーと対話するコンポーネントが設定されます。