An Amazon Managed Streaming for Apache Kafka (Amazon MSK) importtopic lets you continuously ingest data fromAmazon MSK as anexternal source and into Pub/Sub. Then you can stream the datainto any of the destinations that Pub/Sub supports.

This document shows you how to create and manage Amazon MSKimport topics. To create a standard topic, seeCreate a standardtopic.

For more information about import topics, seeAbout import topics.

Before you begin

• Know more about thePub/Sub publishprocess.

• Configure therequired roles and permissions to manage Amazon MSKimport topics including the following:

  • Add the Pub/Sub publisher role to thePub/Sub service account

  • Add the Pub/Sub service agent role to the Pub/Sub service account, if not already granted.

  • Add the service account user role to the service account

• Set upworkload identity federation so thatGoogle Cloud can access the external streaming service.

Required roles and permissions

To get the permissions that you need to create and manage Amazon MSK import topics, ask your administrator to grant you thePub/Sub Editor (roles/pubsub.editor) IAM role on your topic or project. For more information about granting roles, seeManage access to projects, folders, and organizations.

This predefined role contains the permissions required to create and manage Amazon MSK import topics. To see the exact permissions that are required, expand theRequired permissions section:

You might also be able to get these permissions withcustom roles or otherpredefined roles.

You can configure access control at the project level and theindividual resource level.

Set up federated identity to access Amazon MSK

Workload Identity Federation lets Google Cloud services access workloadsrunning outside of Google Cloud. With identity federation, you don't needto maintain or pass credentials to Google Cloud to access your resourcesin other clouds. Instead, you can use the identities of the workloads themselvesto authenticate to Google Cloud and access resources.

Create a service account in Google Cloud

This is an optional step. If you already have a service account, you can useit in this procedure instead of creating a new service account.If you are using an existing service account, go toRecord the service account unique ID for thenext step.

For Amazon MSK import topics, Pub/Sub uses theservice account as the identity to access resources from AWS.

For more information about creating a service account, including prerequisites,required roles and permissions, and naming guidelines, seeCreate service accounts. After you createa service account, you might need to wait for 60 seconds or more before youuse the service account. This behavior occurs because read operations areeventually consistent; it can take time for the new service account tobecome visible.

Record the service account unique ID

You need a service account unique ID to set up a role in the AWS console.

1. In the Google Cloud console, go to theService account details page.

   Go to service account

2. Click the service account that you just created or the one that you areplanning to use.

3. From theService account details page, record the Unique ID number.

   You need the ID as part of the workflow to set upa role in the AWS console.

Add the service account token creator role to the Pub/Sub service account

TheService account token creator role (roles/iam.serviceAccountTokenCreator)lets principalscreate short-lived credentialsfor a service account. These tokens or credentials are used to impersonatethe service account.

For more information about service account impersonation, seeService account impersonation.

You can also add thePub/Sub publisher role (roles/pubsub.publisher)during this procedure. For more information about the role and why you are adding it,seeAdd the Pub/Sub publisher role to the Pub/Sub service account.

1. In the Google Cloud console, go to theIAM page.

   Go to IAM

2. Click theInclude Google-provided role grants checkbox.

3. Look for the service account that has the formatservice-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. For this service account, click theEdit Principal button.

5. If required, clickAdd another role.

6. Search and click theService account token creator role (roles/iam.serviceAccountTokenCreator).

7. ClickSave.

Create a policy in AWS

You need a policy in AWS to let Pub/Sub authenticate to AWS sothat Pub/Sub can ingest data from Amazon MSK.

• For more methods and information about how to create a policy in AWS, seeCreating IAM policies.

To create a policy in AWS, perform the following steps:

1. Sign in to the AWS Management Console and open theIAM console.

2. In the navigation pane of the console forIAM,clickAccess Management >Policies.

3. ClickCreate policy.

4. ForClick a service, clickMSK.

5. ForAction allowed, clickRead >GetBootstrapBrokers.

   This action grants permission to get the bootstrap brokers that Pub/Subuses to connect to the MSK cluster.

6. ClickAdd more permissions.

7. ForSelect a service, clickApache Kafka APIs for MSK.

8. ForAction allowed, select the following:

   • List >DescribeTopic

     This action grants permission to allow the Pub/Sub ingestiontopic to get details about the Amazon MSK Kafka topic.

   • Read >ReadData

     This action grants permission to read data from the Amazon MSK Kafka topic.

   • Write >Connect

     This action grants permission to connect and authenticate to theAmazon MSK Kafka cluster.

9. ForResources, specify thecluster ARN (if you wantto restrict the policy to specific clusters, which is recommended).

10. ClickAdd more permissions.

11. ForSelect a service, clickSTS.

12. ForAction allowed, clickWrite >AssumeRoleWithWebIdentity.

    This action grants permission to obtain a set of temporary securitycredentials for Pub/Sub to authenticate to Amazon MSKby using identity federation.

13. ClickNext.

14. Enter a policy name and description.

15. ClickCreate policy.

Create a role in AWS using a custom trust policy

You must create a role in AWS so that Pub/Sub can authenticateto AWS to ingest data from Amazon MSK.

1. Sign in to the AWS Management Console and open theIAM console.

2. In the navigation pane of the console forIAM, clickRoles.

3. ClickCreate role.

4. ForSelect trusted entity, clickCustom trust policy.

5. In theCustom trust policy section, enter or paste the following:

   {"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Federated":"accounts.google.com"},"Action":"sts:AssumeRoleWithWebIdentity","Condition":{"StringEquals":{"accounts.google.com:sub":"<SERVICE_ACCOUNT_UNIQUE_ID>"}}}]}

   Replace<SERVICE_ACCOUNT_UNIQUE_ID> with the unique ID ofthe service account that you recorded inRecord the service account unique ID.

6. ClickNext.

7. ForAdd permissions, search and click the custom policy thatyou just created.

8. ClickNext.

9. Enter a role name and description.

10. ClickCreate role.

Add the Pub/Sub publisher role to the Pub/Sub principal

To enable publishing, you must assign a publisher role to thePub/Sub service account so that Pub/Sub is able topublish to the Amazon MSK import topic.

Add the Pub/Sub service agent role to the Pub/Sub service account

To allow Pub/Sub to use your import topic project's publish quota, the Pub/Sub service agent requires theserviceusage.services.use permission on your import topic's project.

To provide this permission, we recommend you add the Pub/Sub service agent role to the Pub/Sub service account.

If the Pub/Sub service account does not have the Pub/Sub service agent role, it can be granted as follows:

1. In the Google Cloud console, go to theIAM page.

   Go to IAM

2. Click theInclude Google-provided role grants checkbox.

3. Look for the service account that has the formatservice-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. For this service account, click theEdit principal button.

5. If required, clickAdd another role.

6. Search and click thePub/Sub Service Agent role (roles/pubsub.serviceAgent).

7. ClickSave.

Enable publishing from all topics

Use this method if you have not created any Amazon MSK importtopics.

1. In the Google Cloud console, go to theIAM page.

   Go to IAM

2. Click theInclude Google-provided role grants checkbox.

3. Look for the service account that has the formatservice-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. For this service account, click theEdit principal button.

5. If required, clickAdd another role.

6. Search and click thePub/Sub publisher role (roles/pubsub.publisher).

7. ClickSave.

Enable publishing from a single topic

Use this method only if the Amazon MSK import topic already exists.

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, aCloud 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.

2. Run thegcloud pubsub topics add-iam-policy-binding command:

   gcloudpubsubtopicsadd-iam-policy-bindingTOPIC_ID\--member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com"\--role="roles/pubsub.publisher"

   Replace the following:

   • TOPIC_ID: the topic ID of the Amazon MSK import topic.

   • PROJECT_NUMBER: the project number. To view the projectnumber, seeIdentifying projects.

Add the service account user role to the service account

TheService Account User role (roles/iam.serviceAccountUser) includes thepermissioniam.serviceAccounts.actAs that lets a principal attach a serviceaccount to the Amazon MSK import topic's ingestion settings anduse that service account for federated identity.

1. In the Google Cloud console, go to theIAM page.

   Go to IAM

2. For the principal that's issuing the create or update topic calls, click theEdit principal button.

3. If required, clickAdd another role.

4. Search and click theService account user role(roles/iam.serviceAccountUser).

5. ClickSave.

Use Amazon MSK import topics

You can create a new import topic or edit an existing topic.

Considerations

• Creating the topic and subscription separately, even if done in rapidsuccession, can lead to data loss. There's a short window where the topicexists without a subscription. If any data is sent to the topic during thistime, it is lost. By creating the topic first, creating the subscription,and then converting the topic to an import topic, you guarantee that nomessages are missed during the import process.

• If you need to re-create the Kafka topic of an existing import topic withthe same name, you can'tjust delete the Kafka topic and re-create it.This action can invalidate Pub/Sub's offset management, which can lead todata loss. To mitigate this, follow these steps:

  • Delete the Pub/Sub import topic.
  • Delete the Kafka topic.
  • Create the Kafka topic.
  • Create the Pub/Sub import topic.

• Data from an Amazon MSK Kafka topic is always read from theearliestoffset.

Create Amazon MSK import topics

To know more about properties associated with a topic, seeProperties of atopic.

Ensure that you have completed the following procedures:

• Set up federated identity to access Amazon MSK

• Add the Pub/Sub publisher role to the Pub/Subservice account

• Add the service account user role to the service account

To create an Amazon MSK import topic, follow these steps:

namespacepubsub=::google::cloud::pubsub;namespacepubsub_admin=::google::cloud::pubsub_admin;[](pubsub_admin::TopicAdminClientclient,std::stringproject_id,std::stringtopic_id,std::stringconst&cluster_arn,std::stringconst&msk_topic,std::stringconst&aws_role_arn,std::stringconst&gcp_service_account){google::pubsub::v1::Topicrequest;request.set_name(pubsub::Topic(std::move(project_id),std::move(topic_id)).FullName());auto*aws_msk=request.mutable_ingestion_data_source_settings()->mutable_aws_msk();aws_msk->set_cluster_arn(cluster_arn);aws_msk->set_topic(msk_topic);aws_msk->set_aws_role_arn(aws_role_arn);aws_msk->set_gcp_service_account(gcp_service_account);autotopic=client.CreateTopic(request);// Note that kAlreadyExists is a possible error when the library retries.if(topic.status().code()==google::cloud::StatusCode::kAlreadyExists){std::cout <<"The topic already exists\n";return;}if(!topic)throwstd::move(topic).status();std::cout <<"The topic was successfully created: " <<topic->DebugString()            <<"\n";}

import("context""fmt""io""cloud.google.com/go/pubsub/v2""cloud.google.com/go/pubsub/v2/apiv1/pubsubpb")funccreateTopicWithAWSMSKIngestion(wio.Writer,projectID,topicID,clusterARN,mskTopic,awsRoleARN,gcpSAstring)error{// projectID := "my-project-id"// topicID := "my-topic"// // AWS MSK ingestion settings.// clusterARN := "cluster-arn"// mskTopic := "msk-topic"// awsRoleARN := "aws-role-arn"// gcpSA := "gcp-service-account"ctx:=context.Background()client,err:=pubsub.NewClient(ctx,projectID)iferr!=nil{returnfmt.Errorf("pubsub.NewClient: %w",err)}deferclient.Close()topicpb:=&pubsubpb.Topic{Name:fmt.Sprintf("projects/%s/topics/%s",projectID,topicID),IngestionDataSourceSettings:&pubsubpb.IngestionDataSourceSettings{Source:&pubsubpb.IngestionDataSourceSettings_AwsMsk_{AwsMsk:&pubsubpb.IngestionDataSourceSettings_AwsMsk{ClusterArn:clusterARN,Topic:mskTopic,AwsRoleArn:awsRoleARN,GcpServiceAccount:gcpSA,},},},}topic,err:=client.TopicAdminClient.CreateTopic(ctx,topicpb)iferr!=nil{returnfmt.Errorf("CreateTopic: %w",err)}fmt.Fprintf(w,"Created topic with AWS MSK ingestion settings: %v\n",topic)returnnil}

importcom.google.cloud.pubsub.v1.TopicAdminClient;importcom.google.pubsub.v1.IngestionDataSourceSettings;importcom.google.pubsub.v1.Topic;importcom.google.pubsub.v1.TopicName;importjava.io.IOException;publicclassCreateTopicWithAwsMskIngestionExample{publicstaticvoidmain(String...args)throwsException{// TODO(developer): Replace these variables before running the sample.StringprojectId="your-project-id";StringtopicId="your-topic-id";// AWS MSK ingestion settings.StringclusterArn="cluster-arn";StringmskTopic="msk-topic";StringawsRoleArn="aws-role-arn";StringgcpServiceAccount="gcp-service-account";createTopicWithAwsMskIngestionExample(projectId,topicId,clusterArn,mskTopic,awsRoleArn,gcpServiceAccount);}publicstaticvoidcreateTopicWithAwsMskIngestionExample(StringprojectId,StringtopicId,StringclusterArn,StringmskTopic,StringawsRoleArn,StringgcpServiceAccount)throwsIOException{try(TopicAdminClienttopicAdminClient=TopicAdminClient.create()){TopicNametopicName=TopicName.of(projectId,topicId);IngestionDataSourceSettings.AwsMskawsMsk=IngestionDataSourceSettings.AwsMsk.newBuilder().setClusterArn(clusterArn).setTopic(mskTopic).setAwsRoleArn(awsRoleArn).setGcpServiceAccount(gcpServiceAccount).build();IngestionDataSourceSettingsingestionDataSourceSettings=IngestionDataSourceSettings.newBuilder().setAwsMsk(awsMsk).build();Topictopic=topicAdminClient.createTopic(Topic.newBuilder().setName(topicName.toString()).setIngestionDataSourceSettings(ingestionDataSourceSettings).build());System.out.println("Created topic with AWS MSK ingestion settings: "+topic.getAllFields());}}}

/***TODO(developer):Uncommentthesevariablesbeforerunningthesample.*///consttopicNameOrId='YOUR_TOPIC_NAME_OR_ID';//constclusterArn='arn:aws:kafka:...';//constmskTopic='YOUR_MSK_TOPIC';//constroleArn='arn:aws:iam:...';//constgcpServiceAccount='ingestion-account@...';//ImportstheGoogleCloudclientlibraryconst{PubSub}=require('@google-cloud/pubsub');//Createsaclient;cachethisforfurtheruseconstpubSubClient=newPubSub();asyncfunctioncreateTopicWithAwsMskIngestion(topicNameOrId,clusterArn,mskTopic,awsRoleArn,gcpServiceAccount,){//CreatesanewtopicwithAWSMSKingestion.awaitpubSubClient.createTopic({name:topicNameOrId,ingestionDataSourceSettings:{awsMsk:{clusterArn,topic:mskTopic,awsRoleArn,gcpServiceAccount,},},});console.log(`Topic${topicNameOrId}createdwithAWSMSKingestion.`);}

/***TODO(developer):Uncommentthesevariablesbeforerunningthesample.*///consttopicNameOrId='YOUR_TOPIC_NAME_OR_ID';//constclusterArn='arn:aws:kafka:...';//constmskTopic='YOUR_MSK_TOPIC';//constroleArn='arn:aws:iam:...';//constgcpServiceAccount='ingestion-account@...';//ImportstheGoogleCloudclientlibraryimport{PubSub}from'@google-cloud/pubsub';//Createsaclient;cachethisforfurtheruseconstpubSubClient=newPubSub();asyncfunctioncreateTopicWithAwsMskIngestion(topicNameOrId:string,clusterArn:string,mskTopic:string,awsRoleArn:string,gcpServiceAccount:string,){//CreatesanewtopicwithAWSMSKingestion.awaitpubSubClient.createTopic({name:topicNameOrId,ingestionDataSourceSettings:{awsMsk:{clusterArn,topic:mskTopic,awsRoleArn,gcpServiceAccount,},},});console.log(`Topic${topicNameOrId}createdwithAWSMSKingestion.`);}

fromgoogle.cloudimportpubsub_v1fromgoogle.pubsub_v1.typesimportTopicfromgoogle.pubsub_v1.typesimportIngestionDataSourceSettings# TODO(developer)# project_id = "your-project-id"# topic_id = "your-topic-id"# cluster_arn = "your-cluster-arn"# msk_topic = "your-msk-topic"# aws_role_arn = "your-aws-role-arn"# gcp_service_account = "your-gcp-service-account"publisher=pubsub_v1.PublisherClient()topic_path=publisher.topic_path(project_id,topic_id)request=Topic(name=topic_path,ingestion_data_source_settings=IngestionDataSourceSettings(aws_msk=IngestionDataSourceSettings.AwsMsk(cluster_arn=cluster_arn,topic=msk_topic,aws_role_arn=aws_role_arn,gcp_service_account=gcp_service_account,)),)topic=publisher.create_topic(request=request)print(f"Created topic:{topic.name} with AWS MSK Ingestion Settings")

For more information about ARNs, seeAmazon Resource Names (ARNs) andIAM Identifiers.

If you run into issues, seeTroubleshooting an Amazon MSKimport topic.

Edit Amazon MSK import topics

To edit the ingestion data source settings of an Amazon MSKimport topic, follow these steps:

Quotas and limits

The publisher throughput for import topics is bound by the publish quotaof the topic. For more information, seePub/Sub quotas and limits.

What's next

• Monitor an import topic.

• Choose thetype of subscription for your topic.

• Learn how topublish a message to a topic.

• Create or modify a topic withgcloud CLI,REST APIs, orClient libraries.

• Troubleshoot an Amazon MSK import topic