创建 Amazon Kinesis Data Streams 导入主题

借助 Amazon Kinesis Data Streams 导入主题，您可以持续从 Amazon Kinesis Data Streams（作为外部来源）注入数据，并将其注入到 Pub/Sub 中。然后，您可以将数据流式传输到 Pub/Sub 支持的任何目标位置。

如需详细了解导入主题，请参阅关于导入主题。

准备工作

所需的角色和权限

如需获得创建和管理 Amazon Kinesis Data Streams 导入主题所需的权限，请让您的管理员为您授予主题或项目的 Pub/Sub Editor (roles/pubsub.editor) IAM 角色。 如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

此预定义角色包含创建和管理 Amazon Kinesis Data Streams 导入主题所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

如需创建和管理 Amazon Kinesis Data Streams 导入主题，需要具备以下权限：

  • 创建导入主题： pubsub.topics.create
  • 删除导入主题： pubsub.topics.delete
  • 获取导入主题： pubsub.topics.get
  • 列出导入主题： pubsub.topics.list
  • 发布到导入主题： pubsub.topics.publish and pubsub.serviceAgent
  • 更新导入主题： pubsub.topics.update
  • 获取导入主题的 IAM 政策： pubsub.topics.getIamPolicy
  • 为导入主题配置 IAM 政策： pubsub.topics.setIamPolicy

您也可以使用自定义角色或其他预定义角色来获取这些权限。

您可以在项目级层和个别资源级层配置访问权限控制。

设置联合身份以访问 Amazon Kinesis Data Streams

借助工作负载身份联合， Trusted Cloud 服务可以访问在 Trusted Cloud外部运行的工作负载。借助身份联合，您无需维护凭据或将凭据传递给 Trusted Cloud ，即可访问其他云中的资源。您可以改为使用工作负载本身的身份向 Trusted Cloud 进行身份验证并访问资源。

在 Trusted Cloud中创建服务账号

这是一个可选步骤。 如果您已有服务账号，则可以在此过程中使用该账号，而不必创建新的服务账号。 如果您使用的是现有服务账号，请前往记录服务账号的唯一 ID 以进行下一步。

对于 Amazon Kinesis Data Streams 导入主题，Pub/Sub 使用服务账号作为身份来访问 AWS 中的资源。

如需详细了解如何创建服务账号，包括前提条件、所需角色和权限以及命名指南，请参阅创建服务账号。创建服务账号后，您可能需要等待 60 秒或更长时间才能使用该服务账号。出现这种行为的原因是读取操作是最终一致的；新服务账号可能需要一段时间才能显示。

记录服务账号唯一 ID

您需要服务账号唯一 ID 才能在 AWS 中设置角色。

  1. 在 Trusted Cloud 控制台中，前往服务账号详情页面。

    前往服务账号

  2. 点击您刚刚创建的服务账号或计划使用的服务账号。

  3. 在服务账号详情页面中，记录唯一 ID 编号。

    您需要此 ID 作为工作流的一部分，以便在 AWS 中设置角色。

向 Pub/Sub 服务账号添加服务账号令牌创建者角色

借助 Service Account Token Creator 角色 (roles/iam.serviceAccountTokenCreator)，主账号可以为服务账号创建短期有效凭据。这些令牌或凭据用于模拟服务账号。

如需详细了解服务账号模拟，请参阅服务账号模拟。

您还可以在此过程中添加 Pub/Sub 发布者角色 (roles/pubsub.publisher)。如需详细了解该角色以及您添加该角色的原因，请参阅向 Pub/Sub 服务账号添加 Pub/Sub 发布者角色。

  1. 在 Trusted Cloud 控制台中，前往 IAM 页面。

    转到 IAM

  2. 点击包括 S3NS提供的角色授权复选框。

  3. 查找格式为 service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com 的服务账号。

  4. 对于此服务账号，点击修改主账号按钮。

  5. 如果需要，请点击添加其他角色。

  6. 搜索并点击 Service Account Token Creator 角色 (roles/iam.serviceAccountTokenCreator)。

  7. 点击保存。

在 AWS 中创建政策

您需要在 AWS 中添加一项政策，以允许 Pub/Sub 向 AWS 进行身份验证，这样 Pub/Sub 才能从 Amazon Kinesis Data Streams 注入数据。

  • 如需了解更多方法以及如何在 AWS 中创建政策，请参阅创建 IAM 政策。

如需在 AWS 中创建政策，请执行以下步骤：

  1. 登录 AWS 管理控制台并打开 IAM 控制台。

  2. 在 IAM 控制台的导航窗格中，依次点击访问权限管理 > 政策。

  3. 点击创建政策。

  4. 在选择服务中，点击 Kinesis。

  5. 对于允许的操作，请点击以下内容：

    • List > ListShards。

      此操作会授予列出流中分片的权限，并提供有关每个分片的信息。

    • 读取 > SubscribeToShard。

      此操作可授予权限，以监听具有增强型扇出的特定分片。

    • 读取 > DescribeStreamConsumer。

      此操作授予获取已注册的流消费者的说明的权限。

    这些权限涵盖从流中读取数据。Pub/Sub 仅支持使用流式 SubscribeToShard API 从具有增强型扇出功能的 Kinesis 流中读取数据。

  6. 对于资源，如果您想将政策限制为特定数据流或使用方（推荐），请指定使用方 ARN 和数据流 ARN。

  7. 点击添加更多权限。

  8. 在选择服务中，点击 STS。

  9. 对于允许的操作，依次点击写入 > AssumeRoleWithWebIdentity。

    此操作会授予权限，以获取一组 Pub/Sub 的临时安全凭据，从而通过身份联合向 Amazon Kinesis Data Streams 进行身份验证。

  10. 点击下一步。

  11. 输入政策名称和说明。

  12. 点击创建政策。

使用自定义信任政策在 AWS 中创建角色

您必须在 AWS 中创建一个角色，以便 Pub/Sub 可以向 AWS 进行身份验证，从而从 Amazon Kinesis Data Streams 注入数据。

  1. 登录 AWS 管理控制台并打开 IAM 控制台。

  2. 在 IAM 控制台的导航窗格中，点击 Roles。

  3. 点击 Create role。

  4. 对于选择可信实体，点击自定义信任政策。

  5. 在自定义信任政策部分中，输入或粘贴以下内容：

    {
  "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>"
         }
      }
    }
  ]
}

    将 <SERVICE_ACCOUNT_UNIQUE_ID> 替换为您在记录服务账号的唯一 ID 中记录的服务账号的唯一 ID。

  6. 点击下一步。

  7. 对于添加权限，请搜索并点击您刚刚创建的自定义政策。

  8. 点击下一步。

  9. 输入角色名称和说明。

  10. 点击 Create role。

向 Pub/Sub 主账号添加 Pub/Sub 发布者角色

如需启用发布功能，您必须为 Pub/Sub 服务账号分配发布者角色，以便 Pub/Sub 能够向 Amazon Kinesis Data Streams 导入主题发布内容。

向 Pub/Sub 服务账号添加 Pub/Sub 服务代理角色

为了允许 Pub/Sub 使用导入主题项目的发布配额，Pub/Sub 服务代理需要对导入主题项目拥有 serviceusage.services.use 权限。

如需提供此权限，建议您向 Pub/Sub 服务账号添加 Pub/Sub 服务代理角色。

如果 Pub/Sub 服务账号没有 Pub/Sub 服务代理角色，可以按如下方式授予：

  1. 在 Trusted Cloud 控制台中，前往 IAM 页面。

    转到 IAM

  2. 点击包括 S3NS提供的角色授权复选框。

  3. 查找格式为 service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com 的服务账号。

  4. 对于此服务账号，点击修改主账号按钮。

  5. 如果需要，请点击添加其他角色。

  6. 搜索并点击 Pub/Sub 服务代理角色 (roles/pubsub.serviceAgent)。

  7. 点击保存。

允许发布所有主题的内容

如果您尚未创建任何 Amazon Kinesis Data Streams 导入主题，请使用此方法。

  1. 在 Trusted Cloud 控制台中，前往 IAM 页面。

    转到 IAM

  2. 点击包括 S3NS提供的角色授权复选框。

  3. 查找格式为 service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com 的服务账号。

  4. 对于此服务账号，点击修改主账号按钮。

  5. 如果需要，请点击添加其他角色。

  6. 搜索并点击 Pub/Sub 发布者角色 (roles/pubsub.publisher)。

  7. 点击保存。

启用从单个主题发布内容

仅当 Amazon Kinesis Data Streams 导入主题已存在时，才使用此方法。

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

    Activate Cloud Shell

    At the bottom of the Trusted 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.

  2. 运行 gcloud pubsub topics add-iam-policy-binding 命令：

    gcloud pubsub topics add-iam-policy-binding TOPIC_ID \
   --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com" \
   --role="roles/pubsub.publisher"

    替换以下内容：

    • TOPIC_ID：Amazon Kinesis 数据流导入主题的主题 ID。

    • PROJECT_NUMBER：项目编号。如需查看项目编号，请参阅标识项目。

    3. 向服务账号添加服务账号用户角色

    Service Account User 角色 (roles/iam.serviceAccountUser) 包含权限 iam.serviceAccounts.actAs，该权限可让主账号将服务账号附加到 Amazon Kinesis Data Streams 导入主题的提取设置，并使用该服务账号进行联合身份验证。

    1. 在 Trusted Cloud 控制台中，前往 IAM 页面。

      转到 IAM

    2. 对于发出创建或更新主题调用请求的主账号，请点击修改主账号按钮。

    3. 如果需要，请点击添加其他角色。

    4. 搜索并点击 Service account user 角色 (roles/iam.serviceAccountUser)。

    5. 点击保存。

    使用 Amazon Kinesis Data Streams 主题

    您可以创建新的导入主题，也可以修改现有主题。

    注意事项

    • 即使快速连续地创建主题和订阅，也可能会导致数据丢失。在没有订阅的情况下，主题存在的时间很短。如果在此期间向主题发送任何数据，这些数据都会丢失。通过先创建主题、创建订阅，然后将主题转换为导入主题，您可以确保在导入过程中不会遗漏任何消息。

    创建 Amazon Kinesis Data Streams 导入主题

    如需详细了解与主题关联的属性，请参阅主题的属性。

    确保您已完成以下步骤：

    如需创建 Amazon Kinesis Data Streams 导入主题，请按以下步骤操作：

    控制台

    1. 在 Trusted Cloud 控制台中，前往主题页面。

      打开“主题”。

    2. 点击创建主题。

    3. 在主题 ID 字段中，输入 Amazon Kinesis Data Streams 导入主题的 ID。

      如需详细了解如何命名主题，请参阅命名准则。

    4. 选择添加默认订阅。

    5. 选择启用注入。

    6. 对于提取来源，请选择 Amazon Kinesis Data Streams。

    7. 输入以下详细信息：

      • Kinesis 数据流 ARN：您计划注入到 Pub/Sub 中的 Kinesis 数据流的 ARN。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}。

      • Kinesis 使用方 ARN：已注册到 AWS Kinesis Data Stream 的使用方资源的 ARN。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}。

      • AWS 角色 ARN：AWS 角色的 ARN。角色的 ARN 格式如下： arn:aws:iam::${Account}:role/${RoleName}

      • 服务账号：您在在 Trusted Cloud中创建服务账号中创建的服务账号。

    8. 点击创建主题。

    gcloud

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

      Activate Cloud Shell

      At the bottom of the Trusted 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.

    2. 运行 gcloud pubsub topics create 命令：

      gcloud pubsub topics create TOPIC_ID \
    --kinesis-ingestion-stream-arn KINESIS_STREAM_ARN \
    --kinesis-ingestion-consumer-arn KINESIS_CONSUMER_ARN \
    --kinesis-ingestion-role-arn KINESIS_ROLE_ARN \
    --kinesis-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

      替换以下内容：

      • TOPIC_ID：主题 ID。
      • KINESIS_STREAM_ARN：您计划注入到 Pub/Sub 中的 Kinesis Data Streams 的 ARN。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}。
      • KINESIS_CONSUMER_ARN：已向 AWS Kinesis Data Streams 注册的使用方资源的 ARN。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}。
      • KINESIS_ROLE_ARN：AWS 角色的 ARN。角色的 ARN 格式如下：arn:aws:iam::${Account}:role/${RoleName}。
      • PUBSUB_SERVICE_ACCOUNT：您在 Google Cloud 中创建服务账号中创建的服务账号。

      3. C++

      在尝试此示例之前，请按照《快速入门：使用客户端库》中的 C++ 设置说明进行操作。如需了解详情，请参阅 Pub/Sub C++ API 参考文档。

      namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string stream_arn, std::string consumer_arn,
   std::string aws_role_arn, std::string gcp_service_account) {
  google::pubsub::v1::Topic request;
  request.set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* aws_kinesis =
      request.mutable_ingestion_data_source_settings()->mutable_aws_kinesis();
  aws_kinesis->set_stream_arn(stream_arn);
  aws_kinesis->set_consumer_arn(consumer_arn);
  aws_kinesis->set_aws_role_arn(aws_role_arn);
  aws_kinesis->set_gcp_service_account(gcp_service_account);

  auto topic = 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) throw std::move(topic).status();

  std::cout << "The topic was successfully created: " << topic->DebugString()
            << "\n";
}

      Go

      以下示例使用 Go Pub/Sub 客户端库的主要版本 (v2)。如果您仍在使用 v1 库，请参阅迁移到 v2 的指南。如需查看 v1 代码示例的列表，请参阅 已弃用的代码示例。

      在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Go 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Go API 参考文档。

      import (
	"context"
	"fmt"
	"io"

	pubsub "cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

func createTopicWithKinesisIngestion(w io.Writer, projectID, topic string) error {
	// projectID := "my-project-id"
	// topicID := "projects/my-project-id/topics/my-topic"
	streamARN := "stream-arn"
	consumerARN := "consumer-arn"
	awsRoleARN := "aws-role-arn"
	gcpServiceAccount := "gcp-service-account"

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	topicpb := &pubsubpb.Topic{
		Name: topic,
		IngestionDataSourceSettings: &pubsubpb.IngestionDataSourceSettings{
			Source: &pubsubpb.IngestionDataSourceSettings_AwsKinesis_{
				AwsKinesis: &pubsubpb.IngestionDataSourceSettings_AwsKinesis{
					StreamArn:         streamARN,
					ConsumerArn:       consumerARN,
					AwsRoleArn:        awsRoleARN,
					GcpServiceAccount: gcpServiceAccount,
				},
			},
		},
	}
	topicpb, err = client.TopicAdminClient.CreateTopic(ctx, topicpb)
	if err != nil {
		return fmt.Errorf("failed to create topic with kinesis: %w", err)
	}
	fmt.Fprintf(w, "Kinesis topic created: %v\n", topicpb)
	return nil
}

      Java

      在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Java 设置说明进行操作。 如需了解详情，请参阅 Pub/Sub Java API 参考文档。

      
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

public class CreateTopicWithKinesisIngestionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    // Kinesis ingestion settings.
    String streamArn = "stream-arn";
    String consumerArn = "consumer-arn";
    String awsRoleArn = "aws-role-arn";
    String gcpServiceAccount = "gcp-service-account";

    createTopicWithKinesisIngestionExample(
        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
  }

  public static void createTopicWithKinesisIngestionExample(
      String projectId,
      String topicId,
      String streamArn,
      String consumerArn,
      String awsRoleArn,
      String gcpServiceAccount)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);

      IngestionDataSourceSettings.AwsKinesis awsKinesis =
          IngestionDataSourceSettings.AwsKinesis.newBuilder()
              .setStreamArn(streamArn)
              .setConsumerArn(consumerArn)
              .setAwsRoleArn(awsRoleArn)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();

      Topic topic =
          topicAdminClient.createTopic(
              Topic.newBuilder()
                  .setName(topicName.toString())
                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
                  .build());

      System.out.println("Created topic with Kinesis ingestion settings: " + topic.getAllFields());
    }
  }
}

      Node.js

      在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Node.js 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Node.js API 参考文档。

      /**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithKinesisIngestion(
  topicNameOrId,
  awsRoleArn,
  gcpServiceAccount,
  streamArn,
  consumerArn,
) {
  // Creates a new topic with Kinesis ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

      Node.ts

      在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Node.js 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Node.js API 参考文档。

      /**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithKinesisIngestion(
  topicNameOrId: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
  streamArn: string,
  consumerArn: string,
) {
  // Creates a new topic with Kinesis ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

      Python

      在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Python 设置说明进行操作。 如需了解详情，请参阅 Pub/Sub Python API 参考文档。

      from google.cloud import pubsub_v1
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# stream_arn = "your-stream-arn"
# consumer_arn = "your-consumer-arn"
# 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_kinesis=IngestionDataSourceSettings.AwsKinesis(
            stream_arn=stream_arn,
            consumer_arn=consumer_arn,
            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 Kinesis Ingestion Settings")

    如需详细了解 ARN，请参阅 Amazon 资源名称 (ARN) 和 IAM 标识符。

    如果您遇到问题，请参阅排查 Amazon Kinesis Data Streams 导入主题方面的问题。

    修改 Amazon Kinesis Data Streams 导入主题

    您可以修改 Amazon Kinesis Data Streams 导入主题的提取数据源设置。执行以下步骤：

    控制台

    1. 在 Trusted Cloud 控制台中，前往主题页面。

      转到“主题”

    2. 点击 Amazon Kinesis Data Streams 导入主题。

    3. 在主题详情页面中，点击修改。

    4. 更新您要更改的字段。

    5. 点击更新。

    gcloud

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

      Activate Cloud Shell

      At the bottom of the Trusted 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.

    2. 为避免丢失导入主题的设置，请务必在每次更新主题时包含所有设置。如果您遗漏了某些内容，Pub/Sub 会将相应设置重置为原始默认值。

      运行 gcloud pubsub topics update 命令，并使用以下示例中提及的所有标志：

      gcloud pubsub topics update TOPIC_ID 
      
    --kinesis-ingestion-stream-arn KINESIS_STREAM_ARN 
      
    --kinesis-ingestion-consumer-arn KINESIS_CONSUMER_ARN 
      
    --kinesis-ingestion-role-arn KINESIS_ROLE_ARN 
      
    --kinesis-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

      替换以下内容：

      • TOPIC_ID 是主题 ID。此字段无法更新。

      • KINESIS_STREAM_ARN 是您计划注入到 Pub/Sub 中的 Kinesis Data Streams 的 ARN。ARN 格式如下： arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}。

      • KINESIS_CONSUMER_ARN 是已向 AWS Kinesis Data Streams 注册的使用方资源的 ARN。ARN 格式如下： arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}。

      • KINESIS_ROLE_ARN 是 AWS 角色的 ARN。角色的 ARN 格式如下：arn:aws:iam::${Account}:role/${RoleName}。

      • PUBSUB_SERVICE_ACCOUNT 是您在在 Trusted Cloud中创建服务账号中创建的服务账号。

    Amazon Kinesis Data Streams 导入主题的配额和限制

    导入主题的发布者吞吐量受主题的发布配额限制。如需了解详情，请参阅 Pub/Sub 配额和限制。

    后续步骤

    Apache Kafka® 是 Apache Software Foundation 或其关联公司在美国和/或其他国家/地区的注册商标。