Importthema für Amazon Managed Streaming for Apache Kafka erstellen

Mit einem Importthema für Amazon Managed Streaming for Apache Kafka (Amazon MSK) können Sie Daten kontinuierlich aus Amazon MSK als externe Quelle in Pub/Sub aufnehmen. Anschließend können Sie die Daten in eines der Ziele streamen, die von Pub/Sub unterstützt werden.

In diesem Dokument erfahren Sie, wie Sie Amazon MSK-Importthemen erstellen und verwalten. Informationen zum Erstellen eines Standardthemas finden Sie unter Standardthema erstellen.

Weitere Informationen zu Importthemen finden Sie unter Importthemen.

Hinweise

• Weitere Informationen zum Pub/Sub-Veröffentlichungsprozess

• Konfigurieren Sie die erforderlichen Rollen und Berechtigungen zum Verwalten von Amazon MSK-Importthemen, einschließlich der folgenden:

  • Dem Pub/Sub-Dienstkonto die Rolle „Pub/Sub-Publisher“ hinzufügen

  • Fügen Sie dem Pub/Sub-Dienstkonto die Rolle des Pub/Sub-Dienst-Agents hinzu, falls sie noch nicht zugewiesen ist.

  • Dem Dienstkonto die Rolle „Dienstkontonutzer“ hinzufügen

• Richten Sie die Workload Identity-Föderation ein, damitTrusted Cloud auf den externen Streamingdienst zugreifen kann.

Erforderliche Rollen und Berechtigungen

Bitten Sie Ihren Administrator, Ihnen die IAM-Rolle Pub/Sub Editor (roles/pubsub.editor) für Ihr Thema oder Projekt zuzuweisen, damit Sie die Berechtigungen erhalten, die Sie zum Erstellen und Verwalten von Amazon MSK-Importthemen benötigen. Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Diese vordefinierte Rolle enthält die Berechtigungen, die zum Erstellen und Verwalten von Amazon MSK-Importthemen erforderlich sind. Erweitern Sie den Abschnitt Erforderliche Berechtigungen, um die erforderlichen Berechtigungen anzuzeigen:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind erforderlich, um Amazon MSK-Importthemen zu erstellen und zu verwalten:

• Importthema erstellen: pubsub.topics.create
• Importthema löschen: pubsub.topics.delete
• Importthema abrufen: pubsub.topics.get
• Importthema auflisten: pubsub.topics.list
• In einem Importthema veröffentlichen: pubsub.topics.publish and pubsub.serviceAgent
• Importthema aktualisieren: pubsub.topics.update
• IAM-Richtlinie für ein Importthema abrufen: pubsub.topics.getIamPolicy
• Konfigurieren Sie die IAM-Richtlinie für ein Importthema: pubsub.topics.setIamPolicy

Sie können diese Berechtigungen auch mit benutzerdefinierten Rollen oder anderen vordefinierten Rollen erhalten.

Sie können die Zugriffssteuerung auf Projekt- und auf der Ebene einzelner Ressourcen konfigurieren.

Federated Identity für den Zugriff auf Amazon MSK einrichten

Mit der Workload Identity-Föderation können Trusted Cloud Dienste auf Arbeitslasten zugreifen, die außerhalb von Trusted Cloudausgeführt werden. Bei der Identitätsföderation müssen Sie keine Anmeldedaten für Trusted Cloud verwalten oder übergeben, um auf Ihre Ressourcen in anderen Clouds zuzugreifen. Stattdessen können Sie die Identitäten der Arbeitslasten selbst verwenden, um sich bei Trusted Cloud zu authentifizieren und auf Ressourcen zuzugreifen.

Dienstkonto in Trusted Clouderstellen

Dieser Schritt ist optional. Wenn Sie bereits ein Dienstkonto haben, können Sie es in diesem Verfahren verwenden, anstatt ein neues Dienstkonto zu erstellen. Wenn Sie ein vorhandenes Dienstkonto verwenden, fahren Sie mit dem nächsten Schritt fort: Eindeutige ID des Dienstkontos notieren.

Für Amazon MSK-Importthemen verwendet Pub/Sub das Dienstkonto als Identität für den Zugriff auf Ressourcen von AWS.

Weitere Informationen zum Erstellen eines Dienstkontos, einschließlich Voraussetzungen, erforderlicher Rollen und Berechtigungen sowie Namensrichtlinien, finden Sie unter Dienstkonten erstellen. Nachdem Sie ein Dienstkonto erstellt haben, müssen Sie möglicherweise 60 Sekunden oder länger warten, bis Sie das Dienstkonto verwenden können. Dieses Verhalten tritt auf, weil Lesevorgänge Eventual Consistency haben. Es kann einige Zeit dauern, bis das neue Dienstkonto sichtbar ist.

Eindeutige ID des Dienstkontos notieren

Sie benötigen eine eindeutige Dienstkonto-ID, um eine Rolle in der AWS-Konsole einzurichten.

1. Rufen Sie in der Trusted Cloud Console die Detailseite Dienstkonto auf.

   Zum Dienstkonto

2. Klicken Sie auf das Dienstkonto, das Sie gerade erstellt haben oder das Sie verwenden möchten.

3. Notieren Sie sich auf der Seite Dienstkontodetails die eindeutige ID-Nummer.

   Sie benötigen die ID im Workflow, um eine Rolle in der AWS-Konsole einzurichten.

Rolle „Ersteller von Dienstkonto-Tokens“ dem Pub/Sub-Dienstkonto hinzufügen

Mit der Rolle „Ersteller von Dienstkonto-Tokens“ (roles/iam.serviceAccountTokenCreator) können Hauptkonten kurzlebige Anmeldedaten für ein Dienstkonto erstellen. Diese Tokens oder Anmeldedaten werden verwendet, um die Identität des Dienstkontos zu übernehmen.

Weitere Informationen zu Identitätswechsel für Dienstkonten.

Sie können auch die Rolle „Pub/Sub-Publisher“ (roles/pubsub.publisher) hinzufügen. Weitere Informationen zur Rolle und dazu, warum Sie sie hinzufügen, finden Sie unter Pub/Sub-Publisher-Rolle dem Pub/Sub-Dienstkonto hinzufügen.

1. Rufen Sie in der Trusted Cloud Console die Seite IAM auf.

   IAM aufrufen

2. Klicken Sie auf das Kästchen Von S3NSbereitgestellte Rollenzuweisungen einschließen.

3. Suchen Sie nach dem Dienstkonto im Format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Klicken Sie für dieses Dienstkonto auf die Schaltfläche Hauptkonto bearbeiten.

5. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

6. Suchen Sie nach der Rolle „Ersteller von Dienstkonto-Tokens“ (roles/iam.serviceAccountTokenCreator) und klicken Sie darauf.

7. Klicken Sie auf Speichern.

Richtlinie in AWS erstellen

Sie benötigen eine Richtlinie in AWS, damit sich Pub/Sub bei AWS authentifizieren kann und Daten aus Amazon MSK aufnehmen kann.

• Weitere Methoden und Informationen zum Erstellen einer Richtlinie in AWS finden Sie unter IAM-Richtlinien erstellen.

So erstellen Sie eine Richtlinie in AWS:

1. Melden Sie sich in der AWS Management Console an und öffnen Sie die IAM-Konsole.

2. Klicken Sie in der Console im Navigationsbereich für IAM auf Zugriffsverwaltung > Richtlinien.

3. Klicken Sie auf Richtlinie erstellen.

4. Klicken Sie unter Click a service (Auf einen Dienst klicken) auf MSK.

5. Klicken Sie für Action allowed (Aktion zulässig) auf Read > GetBootstrapBrokers.

   Diese Aktion erteilt die Berechtigung, die Bootstrap-Broker abzurufen, die Pub/Sub für die Verbindung zum MSK-Cluster verwendet.

6. Klicken Sie auf Weitere Berechtigungen hinzufügen.

7. Klicken Sie unter Select a service (Dienst auswählen) auf Apache Kafka APIs for MSK (Apache Kafka-APIs für MSK).

8. Wählen Sie für Zulässige Aktion Folgendes aus:

   • Liste > DescribeTopic

     Durch diese Aktion wird die Berechtigung erteilt, Details zum Amazon MSK Kafka-Thema für das Pub/Sub-Aufnahmethema abzurufen.

   • Lesen > ReadData

     Mit dieser Aktion wird die Berechtigung zum Lesen von Daten aus dem Amazon MSK-Kafka-Thema erteilt.

   • Schreiben > Verbinden

     Mit dieser Aktion wird die Berechtigung zum Herstellen einer Verbindung zum Amazon MSK Kafka-Cluster und zur Authentifizierung bei diesem Cluster gewährt.

9. Geben Sie für Ressourcen den Cluster-ARN an, wenn Sie die Richtlinie auf bestimmte Cluster beschränken möchten. Das wird empfohlen.

10. Klicken Sie auf Weitere Berechtigungen hinzufügen.

11. Klicken Sie unter Dienst auswählen auf STS.

12. Klicken Sie für Action allowed (Aktion zulässig) auf Write (Schreiben) > AssumeRoleWithWebIdentity.

    Mit dieser Aktion wird die Berechtigung erteilt, eine Reihe temporärer Sicherheitsanmeldedaten für Pub/Sub abzurufen, um sich mithilfe der Identitätsföderation bei Amazon MSK zu authentifizieren.

13. Klicken Sie auf Weiter.

14. Geben Sie einen Richtliniennamen und eine Beschreibung ein.

15. Klicken Sie auf Richtlinie erstellen.

Rolle in AWS mit einer benutzerdefinierten Trust-Richtlinie erstellen

Sie müssen eine Rolle in AWS erstellen, damit Pub/Sub sich bei AWS authentifizieren kann, um Daten aus Amazon MSK aufzunehmen.

1. Melden Sie sich in der AWS Management Console an und öffnen Sie die IAM-Konsole.

2. Klicken Sie im Navigationsbereich der Console für IAM auf Rollen.

3. Klicken Sie auf Rolle erstellen.

4. Klicken Sie unter Vertrauenswürdige Entität auswählen auf Benutzerdefinierte Vertrauensrichtlinie.

5. Geben Sie im Abschnitt Benutzerdefinierte Vertrauensrichtlinie Folgendes ein oder fügen Sie es ein:

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

   Ersetzen Sie <SERVICE_ACCOUNT_UNIQUE_ID> durch die eindeutige ID des Dienstkontos, die Sie unter Eindeutige ID des Dienstkontos notieren notiert haben.

6. Klicken Sie auf Weiter.

7. Suchen Sie unter Berechtigungen hinzufügen nach der benutzerdefinierten Richtlinie, die Sie gerade erstellt haben, und klicken Sie darauf.

8. Klicken Sie auf Weiter.

9. Geben Sie einen Rollennamen und eine Beschreibung ein.

10. Klicken Sie auf Rolle erstellen.

Pub/Sub-Publisher-Rolle dem Pub/Sub-Principal hinzufügen

Damit die Veröffentlichung möglich ist, müssen Sie dem Pub/Sub-Dienstkonto eine Publisher-Rolle zuweisen, damit Pub/Sub im Amazon MSK-Importthema veröffentlichen kann.

Pub/Sub-Dienst-Agent-Rolle dem Pub/Sub-Dienstkonto hinzufügen

Damit Pub/Sub das Veröffentlichungskontingent des Projekts Ihres Importthemas verwenden kann, benötigt der Pub/Sub-Dienst-Agent die Berechtigung serviceusage.services.use für das Projekt Ihres Importthemas.

Um diese Berechtigung zu erteilen, empfehlen wir, dem Pub/Sub-Dienstkonto die Rolle „Pub/Sub-Dienst-Agent“ hinzuzufügen.

Wenn das Pub/Sub-Dienstkonto nicht über die Rolle „Pub/Sub-Dienst-Agent“ verfügt, kann sie so zugewiesen werden:

1. Rufen Sie in der Trusted Cloud Console die Seite IAM auf.

   IAM aufrufen

2. Klicken Sie auf das Kästchen Von S3NSbereitgestellte Rollenzuweisungen einschließen.

3. Suchen Sie nach dem Dienstkonto im Format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Klicken Sie für dieses Dienstkonto auf die Schaltfläche Hauptkonto bearbeiten.

5. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

6. Suchen Sie nach der Rolle „Pub/Sub-Dienst-Agent“ (roles/pubsub.serviceAgent) und klicken Sie darauf.

7. Klicken Sie auf Speichern.

Veröffentlichung von allen Themen aus aktivieren

Verwenden Sie diese Methode, wenn Sie noch keine Amazon MSK-Importthemen erstellt haben.

1. Rufen Sie in der Trusted Cloud Console die Seite IAM auf.

   IAM aufrufen

2. Klicken Sie auf das Kästchen Von S3NSbereitgestellte Rollenzuweisungen einschließen.

3. Suchen Sie nach dem Dienstkonto im Format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Klicken Sie für dieses Dienstkonto auf die Schaltfläche Hauptkonto bearbeiten.

5. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

6. Suchen Sie nach der Rolle „Pub/Sub-Publisher“ (roles/pubsub.publisher) und klicken Sie darauf.

7. Klicken Sie auf Speichern.

Veröffentlichung aus einem einzelnen Thema aktivieren

Verwenden Sie diese Methode nur, wenn das Amazon MSK-Importthema bereits vorhanden ist.

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. Führen Sie den Befehl gcloud pubsub topics add-iam-policy-binding aus:

   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"

   Ersetzen Sie Folgendes:

   • TOPIC_ID: die Themen-ID des Amazon MSK-Importthemas.

   • PROJECT_NUMBER: die Projektnummer Informationen zum Aufrufen der Projektnummer finden Sie unter Projekte identifizieren.

Rolle „Dienstkontonutzer“ dem Dienstkonto hinzufügen

Die Rolle „Dienstkontonutzer“ (roles/iam.serviceAccountUser) enthält die Berechtigung iam.serviceAccounts.actAs, mit der ein Hauptkonto ein Dienstkonto an die Aufnahmeeinstellungen des Amazon MSK-Importthemas anhängen und dieses Dienstkonto für die föderierte Identität verwenden kann.

1. Rufen Sie in der Trusted Cloud Console die Seite IAM auf.

   IAM aufrufen

2. Klicken Sie für das Hauptkonto, das die Aufrufe zum Erstellen oder Aktualisieren von Themen ausgibt, auf die Schaltfläche Hauptkonto bearbeiten.

3. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

4. Suchen Sie nach der Rolle „Dienstkontonutzer“ (roles/iam.serviceAccountUser) und klicken Sie darauf.

5. Klicken Sie auf Speichern.

Amazon MSK-Importthemen verwenden

Sie können ein neues Importthema erstellen oder ein vorhandenes Thema bearbeiten.

Hinweise

• Wenn Sie das Thema und das Abo separat erstellen, kann es zu Datenverlusten kommen, auch wenn Sie die Schritte schnell nacheinander ausführen. Es gibt ein kurzes Zeitfenster, in dem das Thema ohne Abo verfügbar ist. Wenn während dieser Zeit Daten an das Thema gesendet werden, gehen sie verloren. Wenn Sie zuerst das Thema und dann das Abo erstellen und das Thema anschließend in ein Importthema umwandeln, stellen Sie sicher, dass während des Importvorgangs keine Nachrichten verloren gehen.

• Wenn Sie das Kafka-Thema eines vorhandenen Importthemas mit demselben Namen neu erstellen müssen, können Sie das Kafka-Thema nicht einfach löschen und neu erstellen. Diese Aktion kann die Offsetverwaltung von Pub/Sub ungültig machen, was zu Datenverlust führen kann. So können Sie das Problem beheben:

  • Löschen Sie das Pub/Sub-Importthema.
  • Löschen Sie das Kafka-Thema.
  • Erstellen Sie das Kafka-Thema.
  • Pub/Sub-Importthema erstellen

• Daten aus einem Amazon MSK-Kafka-Thema werden immer ab dem frühesten Offset gelesen.

Amazon MSK-Importthemen erstellen

Weitere Informationen zu den Attributen eines Themas

Achten Sie darauf, dass Sie die folgenden Schritte ausgeführt haben:

• Föderierte Identität für den Zugriff auf Amazon MSK einrichten

• Dem Pub/Sub-Dienstkonto die Rolle „Pub/Sub-Publisher“ hinzufügen

• Dem Dienstkonto die Rolle „Dienstkontonutzer“ hinzufügen

So erstellen Sie ein Amazon MSK-Importthema:

Console

1. Rufen Sie in der Trusted Cloud Console die Seite Themen auf.

   Themen aufrufen

2. Klicken Sie auf Thema erstellen.

3. Geben Sie im Feld Themen-ID eine ID für das Amazon MSK-Importthema ein. Weitere Informationen zur Benennung von Themen finden Sie in den Benennungsrichtlinien.

4. Wählen Sie Standardabo hinzufügen aus.

5. Wählen Sie Aufnahme aktivieren aus.

6. Wählen Sie als Erfassungsquelle Amazon MSK aus.

7. Geben Sie die folgenden Informationen ein:

   • Cluster-ARN: Der ARN des Amazon MSK-Clusters, aus dem Sie Daten in Pub/Sub aufnehmen. Das ARN-Format ist wie folgt: arn:aws:kafka:${Region}:${Account}:cluster/${ClusterName}/${ClusterId}.
   • Thema: Der Name des Amazon MSK Kafka-Themas, das Sie in Pub/Sub aufnehmen.
   • ARN der AWS-Rolle: Der ARN der AWS-Rolle. Das ARN-Format der Rolle ist arn:aws:iam::${Account}:role/${RoleName}.
   • Dienstkonto: Das Dienstkonto, das Sie in Dienstkonto in Trusted Clouderstellen erstellt haben.

8. Klicken Sie auf Thema erstellen.

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. Führen Sie den Befehl gcloud pubsub topics create aus:

   gcloud pubsub topics create TOPIC_ID \
         --aws-msk-ingestion-cluster-arn MSK_CLUSTER_ARN \
         --aws-msk-ingestion-topic MSK_TOPIC \
         --aws-msk-ingestion-aws-role-arn MSK_ROLE_ARN \
         --aws-msk-ingestion-service-account PUBSUB_SERVICE_ACCOUNT
   

   Ersetzen Sie Folgendes:

   • TOPIC_ID: der Name oder die ID Ihres Pub/Sub-Themas.
   • MSK_CLUSTER_ARN: Der ARN für den Amazon MSK-Cluster, aus dem Sie Daten in Pub/Sub aufnehmen. Das ARN-Format ist wie folgt: arn:aws:kafka:${Region}:${Account}:cluster/${ClusterName}/${ClusterId}.
   • MSK_TOPIC: Der Name des Amazon MSK-Kafka-Themas, das Sie in Pub/Sub aufnehmen.
   • MSK_ROLE_ARN: der ARN der AWS-Rolle. Der ARN der Rolle hat folgendes Format: arn:aws:iam::${Account}:role/${RoleName}.
   • PUBSUB_SERVICE_ACCOUNT: Das Dienstkonto, das Sie in Dienstkonto in Google Cloud erstellen erstellt haben.

C++

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für C++ in der Kurzanleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zur 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 const& cluster_arn,
   std::string const& msk_topic, std::string const& aws_role_arn,
   std::string const& gcp_service_account) {
  google::pubsub::v1::Topic request;
  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);

  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

Im folgenden Beispiel wird die Hauptversion der Go Pub/Sub-Clientbibliothek (v2) verwendet. Wenn Sie noch die v1-Bibliothek verwenden, finden Sie hier den Migrationsleitfaden für v2. Eine Liste der Codebeispiele für Version 1 finden Sie unter Eingestellte Codebeispiele.

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für Go in der Schnellstart-Anleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zu Pub/Sub Go API.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub"
)

func createTopicWithAWSMSKIngestion(w io.Writer, projectID, topicID, clusterARN, mskTopic, awsRoleARN, gcpSA string) 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)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	cfg := &pubsub.TopicConfig{
		IngestionDataSourceSettings: &pubsub.IngestionDataSourceSettings{
			Source: &pubsub.IngestionDataSourceAmazonMSK{
				ClusterARN:        clusterARN,
				Topic:             mskTopic,
				AWSRoleARN:        awsRoleARN,
				GCPServiceAccount: gcpSA,
			},
		},
	}
	t, err := client.CreateTopicWithConfig(ctx, topicID, cfg)
	if err != nil {
		return fmt.Errorf("CreateTopic: %w", err)
	}
	fmt.Fprintf(w, "Created topic with AWS MSK ingestion settings: %v\n", t)
	return nil
}

Java

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für Java in der Kurzanleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zu 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 CreateTopicWithAwsMskIngestionExample {
  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";
    // AWS MSK ingestion settings.
    String clusterArn = "cluster-arn";
    String mskTopic = "msk-topic";
    String awsRoleArn = "aws-role-arn";
    String gcpServiceAccount = "gcp-service-account";

    createTopicWithAwsMskIngestionExample(
        projectId, topicId, clusterArn, mskTopic, awsRoleArn, gcpServiceAccount);
  }

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

      IngestionDataSourceSettings.AwsMsk awsMsk =
          IngestionDataSourceSettings.AwsMsk.newBuilder()
              .setClusterArn(clusterArn)
              .setTopic(mskTopic)
              .setAwsRoleArn(awsRoleArn)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setAwsMsk(awsMsk).build();

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

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

Node.js

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für PHP in der Schnellstart-Anleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zu Pub/Sub Node.js API.

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

// 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 createTopicWithAwsMskIngestion(
  topicNameOrId,
  clusterArn,
  mskTopic,
  awsRoleArn,
  gcpServiceAccount,
) {
  // Creates a new topic with AWS MSK ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsMsk: {
        clusterArn,
        topic: mskTopic,
        awsRoleArn,
        gcpServiceAccount,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS MSK ingestion.`);
}

Node.ts

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für PHP in der Schnellstart-Anleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zu Pub/Sub Node.js API.

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

// 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 createTopicWithAwsMskIngestion(
  topicNameOrId: string,
  clusterArn: string,
  mskTopic: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
) {
  // Creates a new topic with AWS MSK ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsMsk: {
        clusterArn,
        topic: mskTopic,
        awsRoleArn,
        gcpServiceAccount,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS MSK ingestion.`);
}

Python

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für Python in der Schnellstart-Anleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zu 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"
# 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")

Weitere Informationen zu ARNs finden Sie unter Amazon-Ressourcennamen (ARNs) und IAM-Kennungen.

Falls Probleme auftreten, lesen Sie den Abschnitt Fehlerbehebung bei einem Amazon MSK-Importthema.

Amazon MSK-Importthemen bearbeiten

So bearbeiten Sie die Einstellungen der Erfassungsdatenquelle eines Amazon MSK-Importthemas:

Console

1. Rufen Sie in der Trusted Cloud Console die Seite Themen auf.

   Themen aufrufen

2. Klicken Sie auf das Amazon MSK-Importthema.

3. Klicken Sie auf der Seite mit den Themendetails auf Bearbeiten.

4. Aktualisieren Sie die Felder, die Sie ändern möchten.

5. Klicken Sie auf Aktualisieren.

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. Führen Sie den Befehl gcloud pubsub topics update mit allen im folgenden Beispiel genannten Flags aus:

   gcloud pubsub topics update TOPIC_ID \
       --aws-msk-ingestion-cluster-arn MSK_CLUSTER_ARN \
       --aws-msk-ingestion-topic MSK_TOPIC \
       --aws-msk-ingestion-aws-role-arn MSK_ROLE_ARN \
       --aws-msk-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

   Ersetzen Sie Folgendes:

   • TOPIC_ID: der Name oder die ID Ihres Pub/Sub-Themas.
   • MSK_CLUSTER_ARN: Der ARN für den Amazon MSK-Cluster, aus dem Sie Daten in Pub/Sub aufnehmen. Das ARN-Format ist wie folgt: arn:aws:kafka:${Region}:${Account}:cluster/${ClusterName}/${ClusterId}.
   • MSK_TOPIC: der Name des Amazon MSK Kafka-Themas, das Sie in Pub/Sub aufnehmen.
   • MSK_ROLE_ARN: der ARN der AWS-Rolle. Das ARN-Format der Rolle ist: arn:aws:iam::${Account}:role/${RoleName}.
   • PUBSUB_SERVICE_ACCOUNT: Das Dienstkonto, das Sie in Dienstkonto in Google Cloud erstellen erstellt haben.

Kontingente und Limits

Der Publisher-Durchsatz für Importthemen ist an das Veröffentlichungskontingent des Themas gebunden. Weitere Informationen finden Sie unter Pub/Sub-Kontingente und ‑Limits.

Nächste Schritte

• Importthema überwachen

• Wählen Sie den Abotyp für Ihr Thema aus.

• Informationen zum Veröffentlichen einer Nachricht in einem Thema

• Erstellen oder ändern Sie ein Thema mit der gcloud CLI, den REST APIs oder den Clientbibliotheken.

• Fehlerbehebung bei einem Amazon MSK-Importthema

Apache Kafka® ist eine eingetragene Marke der Apache Software Foundation oder ihrer Tochtergesellschaften in den USA und/oder anderen Ländern.