クライアントは、オープンソースの Apache Kafka API を使用して Managed Service for Apache Kafka クラスタに接続できます。すべての接続は TLS を使用して暗号化する必要があります。プレーンテキスト通信はサポートされていません。認証は、SASL または mTLS の 2 つのサポートされているメカニズムのいずれかを使用して処理されます。それぞれ異なる認証情報タイプを使用します。
このドキュメントでは、SASL メソッドを使用して認証を行う方法について説明します。クライアントは、サービス アカウントなどの認可された Identity and Access Management プリンシパルの認証情報を使用して認証します。Managed Service for Apache Kafka は、すべての接続のサーバーサイド ブローカー証明書を管理します。
始める前に
以下の説明をご覧ください。
サービス アカウントにマネージド Kafka クライアント ロールを付与する
クラスタを含むプロジェクトに対する roles/managedkafka.client IAM ロールを、クラスタへの接続に使用するサービス アカウントに付与する必要があります。
マネージド Kafka クライアント ロールには、すべての接続に必要な managedkafka.clusters.connect 権限が含まれています。サービス アカウントにマネージド Kafka クライアント ロールを付与する手順は次のとおりです。
コンソール
- Google Cloud コンソールで、[IAM] ページに移動します。
IAM に移動 - プロジェクトが、Managed Service for Apache Kafka クライアントがアクセスするコンシューマー プロジェクトに設定されていることを確認します。
- [アクセス権を付与] をクリックします。
- 新しいページの [プリンシパルの追加] に、使用しているサービス アカウントのメールアドレスを入力します。
- [ロールを割り当てる] で、[マネージド Kafka クライアント] ロールを選択します。
- [保存] をクリックします。
gcloud CLI
-
In the Google Cloud console, activate Cloud Shell.
At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.
-
gcloud projects add-iam-policy-bindingコマンドを実行します。gcloud projects add-iam-policy-binding PROJECT_ID \ --member serviceAccount:SERVICE_ACCOUNT_EMAIL \ --role roles/managedkafka.client
次のように置き換えます。
-
PROJECT_ID は、プロジェクト ID です。
-
SERVICE_ACCOUNT_EMAIL はサービス アカウントのメールアドレスです。
-
Google Cloudで認証を行うように Kafka クライアントを構成する
次のいずれかのメカニズムを使用して、Kafka クライアントを Google Cloud に対して認証できます。
OAUTHBEARER（推奨）: このメカニズムでは、アプリケーションのデフォルト認証情報（ADC）を使用する必要があります。ADC は、アプリケーション環境に基づいて認証情報を自動的に検索するために認証ライブラリが使用するストラテジです。ADC が認証情報を探す場所と順序については、アプリケーションのデフォルト認証情報の仕組みをご覧ください。
SASL/PLAIN: このメカニズムでは、サービス アカウント キーの JSON ファイルまたはアクセス トークンから派生できるユーザー名とパスワードを使用する必要があります。
通常は、OAUTHBEARER をおすすめします。ただし、SASL/PLAIN はテストに便利なメカニズムです。
OAuthBearer 認証
オープンソースの Kafka API に対する認証方法については、GitHub のドキュメントをご覧ください。
SASL/PLAIN 認証
Managed Service for Apache Kafka は、サービス アカウント キーの JSON ファイルまたはアクセス トークンを使用した SASL/PLAIN 認証をサポートしています。
サービス アカウント キーの JSON ファイル
この方法は、すべての Kafka クライアントに適用できます。
クライアントで使用するサービス アカウントのサービス アカウント キーの JSON ファイルをダウンロードします。
base64-encode を使用してサービス アカウント ファイルをエンコードし、認証文字列として使用します。ファイル名を
my_service_account_key.jsonとします。Linux または macOS システムでは、次のように
base64コマンド（多くの場合デフォルトでインストールされています）を使用します。base64 -w 0 < my_service_account_key.json > password.txtこのコマンドによって、次のアクションが実行されます。
base64 < my_service_account_key.json:my_service_account_key.jsonという名前のファイルの内容を読み取ります。base64 エンコードを使用してファイルの内容をエンコードします。Base64 エンコードは、バイナリデータ（サービス アカウント ファイルの JSON データなど）を ASCII テキストとして表す方法です。これは、テキスト用に設計されたチャネルでデータを送信する場合によく使用されます。
> password.txt:base64コマンドの出力（サービス アカウント ファイルの Base64 エンコード バージョン）をpassword.txtという名前の新しいファイルにリダイレクトします。
次のパラメータを使用して、パスワード ファイルの内容を認証に使用できます。
security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="SERVICE_ACCOUNT_EMAIL_ADDRESS" \ password="CONTENTS_OF_BASE64_ENCODED_PASSWORD_FILE";次のように置き換えます。
SERVICE_ACCOUNT_EMAIL_ADDRESS: 認証に使用するサービス アカウントのメールアドレス。CONTENTS_OF_BASE64_ENCODED_PASSWORD_FILE: 前の手順で取得した Base64 エンコードされたパスワード ファイルの内容。これは 1 行にする必要があります。
クラスタへの受信接続を認証するときに、Managed Service for Apache Kafka は次のことを確認します。
指定されたユーザー名が、パスワードで使用されている鍵のサービス アカウントと一致している。
指定されたサービス アカウント プリンシパルには、クラスタに対する
managedkafka.clusters.connect権限（roles/managedkafka.clientIAM ロールに含まれる）があります。
アクセス トークン
認証に使用するプリンシパルのアクセス トークンを取得します。たとえば、現在の gcloud CLI プリンシパルのアクセス トークンを取得します。
gcloud auth print-access-tokenアクセス トークンは、次のパラメータを使用して認証に使用できます。
security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="PRINCIPAL_EMAIL_ADDRESS" \ password="ACCESS_TOKEN_VALUE";次のように置き換えます。
PRINCIPAL_EMAIL_ADDRESS: アクセス トークンの取得に使用したプリンシパルのメールアドレス。ACCESS_TOKEN_VALUE: 前の手順で取得したアクセス トークンの値。
クラスタへの受信接続を認証するときに、Managed Service for Apache Kafka は次のことを確認します。
アクセス トークンが有効で、期限切れになっていない。
指定されたユーザー名が、アクセス トークンに関連付けられているプリンシパル メールアドレスと一致している。
アクセス トークンのプリンシパルには、クラスタに対する
managedkafka.clusters.connect権限（roles/managedkafka.clientIAM ロールに含まれる）があります。
Workload Identity Federation for GKE
Managed Service for Apache Kafka は、Workload Identity Federation for GKE を使用したオープンソースの Apache Kafka API に対する認証をサポートしています。認証は SASL/PLAIN と SASL/OAUTHBEARER の両方でサポートされています。
Managed Service for Apache Kafka で Workload Identity Federation for GKE を使用するには、次の要件を満たす必要があります。
- GKE バージョン
1.31.1-gke.1241000以降を使用します。 Kubernetes サービス アカウントに
iam.gke.io/return-principal-id-as-email: "true"でアノテーションを付けます。次に例を示します。apiVersion: v1 kind: ServiceAccount metadata: name: kafka-service-account annotations: iam.gke.io/return-principal-id-as-email: "true"ローカル認証サーバーを使用している場合は、google-auth パッケージのバージョン
2.40.3以降も使用していることを確認してください。
GKE プリンシパルに managedkafka.clusters.connect 権限（roles/managedkafka.client IAM ロールに含まれる）があることを確認します。
Managed Service for Apache Kafka は、Fleet ワークロード ID を使用したオープンソースの Apache Kafka API に対する認証をサポートしていません。別の方法として、Kubernetes サービス アカウントを IAM サービス アカウントにリンクすることもできます。
認証エラーのトラブルシューティング
Managed Service for Apache Kafka クライアントが Managed Service for Apache Kafka で認証できない場合は、次のようなエラー メッセージが表示されます。
Exception in thread "main" java.util.concurrent.ExecutionException:
org.apache.kafka.common.errors.SaslAuthenticationException:
Authentication failed: Invalid username or password
この問題を解決するには、次の原因を確認します。
パスワードの形式が正しくなく、base64 でデコードされたときに有効なサービス アカウント キーの JSON BLOB または有効なアクセス トークンを表していない。
認証プリンシパルに、クラスタに対する
managedkafka.clusters.connect権限がありません。指定されたユーザー名が、認証情報のプリンシパルと一致しません。
クライアントが 30 分ごとに頻繁に切断される場合は、クライアントが定期的な再認証をサポートしていないことが原因である可能性があります。Managed Service for Apache Kafka ブローカーでは、connections.max.reauth.ms ブローカー プロパティによって、クライアントが 30 分ごとに再認証を行う必要があります。Kafka クライアント ライブラリのバージョンが 2.2.0 以降で、再認証をサポートしていることを確認します。