Kafka 用 Azure Event Hubs に対してパスワードレス接続を使用するようにアプリケーションを移行する

1. Azure portal で、メインの検索バーまたは左側のナビゲーションを使用して Event Hubs 名前空間を見つけます。

2. Event Hubs の概要ページで、左側のメニューから [アクセス制御 (IAM)] を選択します。

3. [アクセス制御 (IAM)] ページで、[ロールの割り当て] タブを選びます。

4. 上部のメニューから [追加] を選択し、結果のドロップダウン メニューから [ロールの割り当ての追加] を選択します。

   

5. 検索ボックスを使って、結果を目的のロールに絞り込みます。 この例では、Azure Event Hubs Data Sender と Azure Event Hubs Data Receiver を検索し、一致する結果を選択して、[次へ] を選択します。

6. [アクセスの割り当て] で、[ユーザー、グループ、またはサービス プリンシパル] を選択し、[メンバーの選択] を選択します。

7. ダイアログで、自分の Microsoft Entra ユーザー名 (通常は user@domain メール アドレス) を検索し、ダイアログの下部にある [選択] を選びます。

8. [レビューと割り当て] を選んで最終ページに移動し、もう一度 [レビューと割り当て] を行ってプロセスを完了します。

Azure CLI を使用して認証するには、次の手順を使用します。

1. 次のコマンドを使用して、Azure アカウントにサインインします。

   az login
   

2. 現在のサブスクリプション コンテキストを設定するには、次のコマンドを使用します。 Azure で使用するサブスクリプションの GUID に置き換えます ssssssss-ssss-ssss-ssss-ssssssssssss 。

   az account set --subscription ssssssss-ssss-ssss-ssss-ssssssssssss
   

3. 次のコマンドを使用して、Azure Event Hubs 名前空間のリソース ID を取得します。

   export AZURE_EVENTHUBS_RESOURCE_ID=$(az resource show \
       --resource-group $AZ_RESOURCE_GROUP \
       --name $AZ_EVENTHUBS_NAMESPACE_NAME \
       --resource-type Microsoft.EventHub/Namespaces \
       --query "id" \
       --output tsv)
   

4. 次のコマンドを使用して、Azure CLI ユーザー アカウントのユーザー オブジェクト ID を取得します。

   export AZURE_AD_ACCOUNT_ID=$(az ad signed-in-user show \
       --query "id" --output tsv)
   

5. 次のコマンドを使用して、Azure Event Hubs Data Sender ロールと Azure Event Hubs Data Receiver ロールをアカウントに割り当てます。

   az role assignment create \
       --assignee $AZURE_AD_ACCOUNT_ID \
       --role "Azure Event Hubs Data Receiver" \
       --scope $AZURE_EVENTHUBS_RESOURCE_ID
   az role assignment create \
       --assignee $AZURE_AD_ACCOUNT_ID \
       --role "Azure Event Hubs Data Sender" \
       --scope $AZURE_EVENTHUBS_RESOURCE_ID
   

次のコマンドを使用して、Azure CLI を使用して Azure にサインインします。

az login

次のコマンドを使用して、PowerShell を使用して Azure にサインインします。

Connect-AzAccount

Visual Studio の 右上隅にある [サインイン ] ボタンを選択します。

以前にロールを割り当てた Microsoft Entra アカウントを使用してサインインします。

Azure アカウント拡張機能がインストールされていることを確認してください。

Ctrl + Shift + P のキーボード ショートカットを使ってコマンド パレットを開きます。 [Azure: サインイン] コマンドを検索し、プロンプトに従って認証します。 Blob Storage アカウントから以前にロールを割り当てた Microsoft Entra アカウントを必ず使用してください。

詳細については、「Azure Toolkit for IntelliJ のインストール」および「Azure Toolkit for IntelliJ のサインイン手順」を参照してください。

1. プロジェクト内で、pom.xml ファイルを開き、次の参照を追加します。

   <dependency>
      <groupId>com.azure</groupId>
      <artifactId>azure-identity</artifactId>
      <version>1.6.0</version>
   </dependency>
   

2. 移行後、次の例に示すように、OAuth2 認証用に AuthenticateCallbackHandler と OAuthBearerToken をプロジェクトに実装します。

   public class KafkaOAuth2AuthenticateCallbackHandler implements AuthenticateCallbackHandler {
   
      private static final Duration ACCESS_TOKEN_REQUEST_BLOCK_TIME = Duration.ofSeconds(30);
      private static final String TOKEN_AUDIENCE_FORMAT = "%s://%s/.default";
   
      private Function<TokenCredential, Mono<OAuthBearerTokenImp>> resolveToken;
      private final TokenCredential credential = new DefaultAzureCredentialBuilder().build();
   
      @Override
      public void configure(Map<String, ?> configs, String mechanism, List<AppConfigurationEntry> jaasConfigEntries) {
         TokenRequestContext request = buildTokenRequestContext(configs);
         this.resolveToken = tokenCredential -> tokenCredential.getToken(request).map(OAuthBearerTokenImp::new);
      }
   
      private TokenRequestContext buildTokenRequestContext(Map<String, ?> configs) {
         URI uri = buildEventHubsServerUri(configs);
         String tokenAudience = buildTokenAudience(uri);
   
         TokenRequestContext request = new TokenRequestContext();
         request.addScopes(tokenAudience);
         return request;
      }
   
      @SuppressWarnings("unchecked")
      private URI buildEventHubsServerUri(Map<String, ?> configs) {
         String bootstrapServer = Arrays.asList(configs.get(BOOTSTRAP_SERVERS_CONFIG)).get(0).toString();
         bootstrapServer = bootstrapServer.replaceAll("\\[|\\]", "");
         URI uri = URI.create("https://" + bootstrapServer);
         return uri;
      }
   
      private String buildTokenAudience(URI uri) {
         return String.format(TOKEN_AUDIENCE_FORMAT, uri.getScheme(), uri.getHost());
      }
   
      @Override
      public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
         for (Callback callback : callbacks) {
            if (callback instanceof OAuthBearerTokenCallback) {
               OAuthBearerTokenCallback oauthCallback = (OAuthBearerTokenCallback) callback;
               this.resolveToken
                       .apply(credential)
                       .doOnNext(oauthCallback::token)
                       .doOnError(throwable -> oauthCallback.error("invalid_grant", throwable.getMessage(), null))
                       .block(ACCESS_TOKEN_REQUEST_BLOCK_TIME);
            } else {
               throw new UnsupportedCallbackException(callback);
            }
         }
      }
   
      @Override
      public void close() {
         // NOOP
      }
   }
   

   public class OAuthBearerTokenImp implements OAuthBearerToken {
       private final AccessToken accessToken;
       private final JWTClaimsSet claims;
   
       public OAuthBearerTokenImp(AccessToken accessToken) {
           this.accessToken = accessToken;
           try {
               claims = JWTParser.parse(accessToken.getToken()).getJWTClaimsSet();
           } catch (ParseException exception) {
               throw new SaslAuthenticationException("Unable to parse the access token", exception);
           }
       }
   
       @Override
       public String value() {
           return accessToken.getToken();
       }
   
       @Override
       public Long startTimeMs() {
           return claims.getIssueTime().getTime();
       }
   
       @Override
       public long lifetimeMs() {
           return claims.getExpirationTime().getTime();
       }
   
       @Override
       public Set<String> scope() {
           // Referring to https://docs.microsoft.com/azure/active-directory/develop/access-tokens#payload-claims, the scp
           // claim is a String, which is presented as a space separated list.
           return Optional.ofNullable(claims.getClaim("scp"))
                   .map(s -> Arrays.stream(((String) s)
                   .split(" "))
                   .collect(Collectors.toSet()))
                   .orElse(null);
       }
   
       @Override
       public String principalName() {
           return (String) claims.getClaim("upn");
       }
   
       public boolean isExpired() {
           return accessToken.isExpired();
       }
   }
   

3. Kafka プロデューサーまたはコンシューマーを作成するときに、SASL/OAUTHBEARER メカニズムをサポートするために必要な構成を追加します。 次の例は、移行前と移行後のコードの外観を示しています。 どちらの例でも、プレースホルダーを <eventhubs-namespace> Event Hubs 名前空間の名前に置き換えます。

   移行前のコードは、次の例のようになります。

   Properties properties = new Properties();
   properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "<eventhubs-namespace>.servicebus.windows.net:9093");
   properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
   properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
   properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
   properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
   properties.put(SaslConfigs.SASL_JAAS_CONFIG,
           String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"%s\";", connectionString));
   return new KafkaProducer<>(properties);
   

   移行後、コードは次の例のようになります。 この例では、プレースホルダーを <path-to-your-KafkaOAuth2AuthenticateCallbackHandler> 実装した KafkaOAuth2AuthenticateCallbackHandlerクラスの完全な名前に置き換えます。

   Properties properties = new Properties();
   properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "<eventhubs-namespace>.servicebus.windows.net:9093");
   properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
   properties.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
   properties.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required");
   properties.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, "<path-to-your-KafkaOAuth2AuthenticateCallbackHandler>");
   return new KafkaProducer<>(properties);
   

1. プロジェクト内で、次の参照をパッケージに com.azure.spring:spring-cloud-azure-starter 追加します。 このライブラリには、パスワードレス接続を実装するために必要なすべてのエンティティが含まれています。

   <dependency>
       <groupId>com.azure.spring</groupId>
       <artifactId>spring-cloud-azure-starter</artifactId>
   </dependency>
   

2. 次に、既存の接続または資格情報に関連するすべての Kafka プロパティを削除します。 次のプロパティのみを追加して、Azure Event Hubs 名前空間で Kafka ブートストラップ サーバーを構成します。

   spring.kafka.bootstrap-servers=$AZ_EVENTHUBS_NAMESPACE_NAME.servicebus.windows.net:9093
   

1. プロジェクト内で、次の参照をパッケージに com.azure.spring:spring-cloud-azure-starter 追加します。 このライブラリには、パスワードレス接続を実装するために必要なすべてのエンティティが含まれています。

   <dependency>
       <groupId>com.azure.spring</groupId>
       <artifactId>spring-cloud-azure-starter</artifactId>
   </dependency>
   

2. 次に、既存の接続または資格情報に関連するすべての Kafka プロパティを削除します。 次のプロパティを追加して、Azure Event Hubs 名前空間を使用して Kafka ブートストラップ サーバーを構成します。

   spring.cloud.stream.kafka.binder.brokers=$AZ_EVENTHUBS_NAMESPACE_NAME.servicebus.windows.net:9093
   

   Note

   バージョン 4.4.0以降では、このプロパティは自動的に追加されるため、手動で追加する必要はありません。

1. Azure アプリ サービス インスタンスのメイン概要ページで、ナビゲーション ウィンドウから [ID] を選択します。

2. [システム割り当て済み] タブで、[ステータス] フィールドをオンに設定します。 システム割り当て ID は Azure によって内部的に管理され、この ID によって管理タスクが自動的に処理されます。 ID の詳細と ID がコードで公開されることはありません。

Service Connector を使用すると、システム割り当てマネージド ID を Azure ホスティング環境に割り当てた後、マネージド ID の Azure Event Hubs Data Sender ロールと Azure Event Hubs Data Receiver ロールを構成するのに役立ちます。

現在、次のコンピューティング サービスがサポートされています。

• Azure App Service
• Azure Spring Apps
• Azure Container Apps

この移行ガイドでは App Service を使用しますが、手順は Azure Spring Apps と Azure Container Apps に似ています。

1. Azure portal の App Service インスタンスのメイン概要ページで、ナビゲーション ウィンドウから Service Connector を選択します。

2. メイン メニューから [作成] を選択すると、[接続の作成] パネルが開きます。 次の値を入力します。

   • サービスの種類: Event Hubs を選択 します。
   • サブスクリプション: 使用するサブスクリプションを選択します。
   • 接続名: connector_appservice_eventhubなど、接続の名前を入力します。
   • 名前空間: 使用する Event Hubs 名前空間を選択します。
   • [クライアントの種類]: 既定値を選択したままにするか、使用する特定のクライアントを選択します。

3. [次へ: 認証] を選択します。

   

4. [システム割り当てマネージド ID (推奨)] が選択されていることを確認し、[次へ: ネットワーク] を選択します。

5. 既定値は選択したままにし、[次へ: 確認と作成] を選択します。

6. Azure で設定が検証された後、[作成] を選択します。

Service Connector は、アプリ サービスのシステム割り当てマネージド ID を自動的に割り当てます。 コネクタでは、選択した Event Hubs インスタンスの Azure Event Hubs データ送信者 と Azure Event Hubs Data Receiver のマネージド ID ロールも割り当てられます。

1. Azure Container Apps インスタンスのメインの概要ページで、ナビゲーション ウィンドウから [ID] を選択します。

2. [システム割り当て済み] タブで、[ステータス] フィールドをオンに設定します。 システム割り当て ID は Azure によって内部的に管理され、この ID によって管理タスクが自動的に処理されます。 ID の詳細と ID がコードで公開されることはありません。

   

1. Azure Spring Apps インスタンスのメイン概要ページで、ナビゲーション ウィンドウから [ID] を選択します。

2. [システム割り当て済み] タブで、[ステータス] フィールドをオンに設定します。 システム割り当て ID は Azure によって内部的に管理され、この ID によって管理タスクが自動的に処理されます。 ID の詳細と ID がコードで公開されることはありません。

   

1. 仮想マシンのメインの概要ページで、ナビゲーション ウィンドウから [ID] を選択します。

2. [システム割り当て済み] タブで、[ステータス] フィールドをオンに設定します。 システム割り当て ID は Azure によって内部的に管理され、この ID によって管理タスクが自動的に処理されます。 ID の詳細と ID がコードで公開されることはありません。

   

Azure Kubernetes Service (AKS) クラスターでは、ロード バランサーやマネージド ディスクなどの Azure リソースにアクセスするための ID が必要です。 この ID には、マネージド ID またはサービス プリンシパルを指定できます。 既定では、AKS クラスターを作成すると、システム割り当てマネージド ID が自動的に作成されます。

次の例に示すように、az webapp identity assign コマンドを使用して、マネージド ID を Azure アプリ Service インスタンスに割り当てることができます。

export AZURE_MANAGED_IDENTITY_ID=$(az webapp identity assign \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <app-service-name> \
    --query principalId \
    --output tsv)

Azure CLI を使用して、Azure コンピューティング ホスティング環境とターゲット サービスの間にサービス接続を作成できます。 「Azure ホスティング環境のマネージド ID の割り当て」セクションで説明されているように、Azure CLI はマネージド ID の作成を自動的に処理し、適切なロールを割り当てます。

まず、Azure CLI 用 Service Connector パスワードレス拡張機能をインストールします。

az extension add --name serviceconnector-passwordless --upgrade

Azure アプリ サービスを使用している場合は、次の例に示すように az webapp connection コマンドを使用します。

az webapp connection create eventhub \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <app-service-name>
    --target-id $AZURE_EVENTHUBS_RESOURCE_ID \
    --client-type kafka-springBoot \
    --system-identity

Azure Spring Apps を使用している場合は、次の az spring connection 例に示すようにコマンドを使用します。

az spring connection create eventhub \
    --app <spring-app-name> \
    --service <azure-spring-service-name> \
    --resource-group $AZ_RESOURCE_GROUP \
    --deployment <spring-app-deployoment-name> \
    --target-id $AZURE_EVENTHUBS_RESOURCE_ID \
    --client-type kafka-springBoot \
    --system-identity

Azure Container Apps を使用している場合は、次の az containerapp connection 例に示すように、コマンドを使用します。

az containerapp connection create eventhub \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <container-app-name>
    --target-id $AZURE_EVENTHUBS_RESOURCE_ID \
    --client-type kafka-springBoot \
    --system-identity

次の例に示すように、az containerapp identity assign コマンドを 使用して、マネージド ID を Azure Container Apps インスタンスに割り当 てることができます。

export AZURE_MANAGED_IDENTITY_ID=$(az containerapp identity assign \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <container-app-name> \
    --system-assigned \
    --query principalId \
    --output tsv)

次の例に示すように、az spring app identity assign コマンドを 使用して、マネージド ID を Azure Spring Apps インスタンスに割り当 てることができます。

export AZURE_MANAGED_IDENTITY_ID=$(az spring app identity assign \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <spring-app-name> \
    --service <spring-apps-service-name> \
    --system-assigned \
    --query identity.principalId \
    --output tsv)

次の例に示すように、az vm identity assign コマンドを 使用して、マネージド ID を仮想マシンに割り当 てることができます。

export AZURE_MANAGED_IDENTITY_ID=$(az vm identity assign \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <vm-name> \
    --query principalId \
    --output tsv)

次の例に示すように、az aks update コマンドを使用して、マネージド ID を Azure Kubernetes Service (AKS) インスタンスに割り当てることができます。

export AZURE_MANAGED_IDENTITY_ID=$(az aks update \
    --resource-group $AZ_RESOURCE_GROUP \
    --name <AKS-cluster-name> \
    --enable-managed-identity \
    --query identityProfile.kubeletidentity.clientId \
    --output tsv)

Service Connector を使用してサービスを接続した場合は、この手順を完了する必要はありません。 次の必要な構成が自動的に処理されました。

• 接続の作成時にマネージド ID を選択した場合、システム割り当てマネージド ID がアプリ用に作成され、Event Hubs 上に Azure Event Hubs データ送信者ロールと Azure Event Hubs データ レシーバー ロールが割り当てられます。

• 接続文字列を使用することを選択した場合、接続文字列はアプリ環境変数として追加されました。

1. Azure portal で、メインの検索バーまたはナビゲーション ウィンドウを使用して Event Hubs 名前空間を見つけます。

2. Event Hubs の概要ページで、ナビゲーション メニューからアクセス制御 (IAM) を選択します。

3. [アクセス制御 (IAM)] ページで、[ロールの割り当て] タブを選びます。

4. メイン メニューから [追加] を選択し、ドロップダウン メニューから [ロールの割り当ての追加] を選択します。

   

5. 検索ボックスを使って、結果を目的のロールに絞り込みます。 この例では、Azure Event Hubs Data Sender と Azure Event Hubs Data Receiver を検索し、一致する結果を選択して、[次へ] を選択します。

6. [アクセスの割り当て] で [マネージド ID] を選択し、[メンバーの選択] を選択します。

7. ポップアップで、ホスティング サービスが配置されているサブスクリプションを検索します。 次に、[ すべてのシステム割り当てマネージド ID] を選択し、ホスティング サービスのマネージド ID を選択します。 システム割り当て ID を選択し、[選択] を選択してポップアップ メニューを閉じます。

   Note

   Azure Kubernetes Service を使用する場合は、[ ユーザー割り当てマネージド ID] を 選択し、次の構造 <your-kubernetes-cluster-name>-agentpoolの名前を持つ Kubernetes クラスターのマネージド ID を選択します。

8. [確認と割り当て] を選択してロールの割り当てを完了できるようになるまで、[次へ] を数回選択します。

Azure CLI を使用してリソース レベルでロールを割り当てるには、次のコマンドを使用できます。

az role assignment create \
    --assignee $AZURE_MANAGED_IDENTITY_ID \
    --role "Azure Event Hubs Data Receiver" \
    --scope $AZURE_EVENTHUBS_RESOURCE_ID
az role assignment create \
    --assignee $AZURE_MANAGED_IDENTITY_ID \
    --role "Azure Event Hubs Data Sender" \
    --scope $AZURE_EVENTHUBS_RESOURCE_ID