Pub/Sub ソース コネクタを作成する

Pub/Sub ソースコネクタは、Pub/Sub から Managed Service for Apache Kafka にメッセージをストリーミングします。これにより、Pub/Sub を Kafka ベースのアプリケーションやデータ パイプラインと統合できます。一般的なユースケースには、ストリーム処理のためのリアルタイム データの取り込みや、イベント ドリブン アーキテクチャの構築などがあります。

概要

このコネクタは、Pub/Sub サブスクリプションからメッセージを読み取り、各メッセージを Kafka レコードに変換して、レコードを Kafka トピックに書き込みます。デフォルトでは、コネクタは次のように Kafka レコードを作成します。

• Kafka レコードキーは null です。
• Kafka レコード値は、Pub/Sub メッセージ データ（バイト単位）です。
• Kafka レコード ヘッダーが空です。

ただし、この動作は構成できます。詳細については、コネクタを構成するをご覧ください。

始める前に

Pub/Sub ソースコネクタを作成する前に、次のものが揃っていることを確認してください。

• サブスクリプションを含む Pub/Sub トピック。

• Managed Service for Apache Kafka クラスタ。

• Kafka クラスタ内の Kafka トピック。

• Connect クラスタ。Connect クラスタを作成するときに、Managed Service for Apache Kafka クラスタをプライマリ Kafka クラスタとして設定します。

必要なロールと権限

コネクタの作成に必要な権限を取得するには、プロジェクトに対するマネージド Kafka コネクタ編集者 （roles/managedkafka.connectorEditor）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

この事前定義ロールには、コネクタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

コネクタを作成するには、次の権限が必要です。

• コネクタを作成します。 managedkafka.connectors.create

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

Pub/Sub から読み取る権限を付与する

マネージド Kafka サービス エージェントには、Pub/Sub サブスクリプションからメッセージを読み取る権限が必要です。サブスクリプションを含むプロジェクトに対する次の IAM ロールをサービス エージェントに付与します。

• Pub/Sub サブスクライバー（roles/pubsub.subscriber）
• Pub/Sub 閲覧者（roles/pubsub.viewer）

Connect クラスタが Managed Service for Apache Kafka クラスタとは異なるプロジェクトにある場合は、 別のプロジェクトに Connect クラスタを作成するをご覧ください。

Pub/Sub ソースコネクタを作成する

コンソール

1. Google Cloud コンソールで、[Connect クラスタ] ページに移動します。

   [Connect クラスタ] に移動

2. コネクタを作成する Connect クラスタをクリックします。

3. [コネクタを作成] をクリックします。

4. コネクタ名に文字列を入力します。

   コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。

5. [コネクタ プラグイン] で [Pub/Sub ソース] を選択します。

6. [Cloud Pub/Sub サブスクリプション] リストで、Pub/Sub サブスクリプションを選択します。コネクタは、このサブスクリプションからメッセージを pull します。サブスクリプションは、完全なリソース名 projects/{project}/subscriptions/{subscription} として表示されます。

7. [Kafka トピック] リストで、メッセージが書き込まれる Kafka トピックを選択します。

8. 省略可: [構成] ボックスで、構成プロパティを追加するか、デフォルトのプロパティを編集します。詳細については、コネクタを構成するをご覧ください。

9. [タスクの再起動ポリシー] を選択します。詳細については、タスクの再起動ポリシーをご覧ください。

10. [作成] をクリックします。

gcloud

gcloud managed-kafka connectors create コマンドを実行します。

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

次のように置き換えます。

• CONNECTOR_ID: コネクタの ID または名前。
• CONNECT_CLUSTER_ID: コネクタが作成される Connect クラスタの ID。
• LOCATION: Connect クラスタのロケーション。
• CONFIG_FILE: ローカルの YAML または JSON 構成ファイルのパス。

以下に示すのは、構成ファイルの例です。

connector.class: "com.google.pubsub.kafka.source.CloudPubSubSourceConnector"
cps.project: "PROJECT_ID"
cps.subscription: "PUBSUB_SUBSCRIPTION_ID"
kafka.topic: "KAFKA_TOPIC_ID"
value.converter: "org.apache.kafka.connect.converters.ByteArrayConverter"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
tasks.max: "3"

次のように置き換えます。

• PROJECT_ID: Pub/Sub サブスクリプションが存在する Google Cloudプロジェクトの ID。

• PUBSUB_SUBSCRIPTION_ID: データを pull する Pub/Sub サブスクリプションの ID。

• KAFKA_TOPIC_ID: データが書き込まれる Kafka トピックの ID。

cps.project、cps.subscription、kafka.topic の構成プロパティは必須です。その他の構成オプションについては、コネクタを構成するをご覧ください。

Terraform

コネクタの作成には Terraform リソースを使用します。

resource "google_managed_kafka_connector" "example-pubsub-source-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-pubsub-source-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "connector.class"  = "com.google.pubsub.kafka.source.CloudPubSubSourceConnector"
    "name"             = "my-pubsub-source-connector"
    "tasks.max"        = "3"
    "kafka.topic"      = "GMK_TOPIC_ID"
    "cps.subscription" = "CPS_SUBSCRIPTION_ID"
    "cps.project"      = data.google_project.default.project_id
    "value.converter"  = "org.apache.kafka.connect.converters.ByteArrayConverter"
    "key.converter"    = "org.apache.kafka.connect.storage.StringConverter"
  }

  provider = google-beta
}

Terraform 構成を適用または削除する方法については、基本的な Terraform コマンドをご覧ください。

Go

このサンプルを試す前に、 クライアント ライブラリをインストールするの Go の設定手順を実施してください。詳細については、 Managed Service for Apache Kafka Go API リファレンス ドキュメントをご覧ください。

Managed Service for Apache Kafka に対する認証を行うには、アプリケーションのデフォルト認証情報（ADC）を設定します。詳細については、ローカル開発環境の ADC の設定をご覧ください。

import (
	"context"
	"fmt"
	"io"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"
)

// createPubSubSourceConnector creates a Pub/Sub Source connector.
func createPubSubSourceConnector(w io.Writer, projectID, region, connectClusterID, connectorID, kafkaTopic, cpsSubscription, cpsProject, tasksMax, valueConverter, keyConverter string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "CPS_SOURCE_CONNECTOR_ID"
	// kafkaTopic := "GMK_TOPIC_ID"
	// cpsSubscription := "CPS_SUBSCRIPTION_ID"
	// cpsProject := "GCP_PROJECT_ID"
	// tasksMax := "3"
	// valueConverter := "org.apache.kafka.connect.converters.ByteArrayConverter"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	// Pub/Sub Source sample connector configuration
	config := map[string]string{
		"connector.class":  "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
		"name":             connectorID,
		"tasks.max":        tasksMax,
		"kafka.topic":      kafkaTopic,
		"cps.subscription": cpsSubscription,
		"cps.project":      cpsProject,
		"value.converter":  valueConverter,
		"key.converter":    keyConverter,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created Pub/Sub source connector: %s\n", resp.Name)
	return nil
}

Java

このサンプルを試す前に、 クライアント ライブラリをインストールするの Java の設定手順を行ってください。詳細については、 Managed Service for Apache Kafka Java API リファレンス ドキュメントをご覧ください。

Managed Service for Apache Kafka に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、 ローカル開発環境の ADC の設定をご覧ください。

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreatePubSubSourceConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-pubsub-source-connector";
    String pubsubProjectId = "my-pubsub-project-id";
    String subscriptionName = "my-subscription";
    String kafkaTopicName = "pubsub-topic";
    String connectorClass = "com.google.pubsub.kafka.source.CloudPubSubSourceConnector";
    String maxTasks = "3";
    String valueConverter = "org.apache.kafka.connect.converters.ByteArrayConverter";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createPubSubSourceConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        pubsubProjectId,
        subscriptionName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        valueConverter,
        keyConverter);
  }

  public static void createPubSubSourceConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String pubsubProjectId,
      String subscriptionName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String valueConverter,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("name", connectorId);
    configMap.put("tasks.max", maxTasks);
    configMap.put("kafka.topic", kafkaTopicName);
    configMap.put("cps.subscription", subscriptionName);
    configMap.put("cps.project", pubsubProjectId);
    configMap.put("value.converter", valueConverter);
    configMap.put("key.converter", keyConverter);

    Connector connector = Connector.newBuilder()
        .setName(
            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
        .putAllConfigs(configMap)
        .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
          .setConnectorId(connectorId)
          .setConnector(connector)
          .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created Pub/Sub Source connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

このサンプルを試す前に、 クライアント ライブラリをインストールするの Python の設定手順を実施してください。詳細については、 Managed Service for Apache Kafka Python API リファレンス ドキュメントをご覧ください。

Managed Service for Apache Kafka に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の ADC の設定をご覧ください。

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "connector.class": "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
    "name": connector_id,
    "tasks.max": tasks_max,
    "kafka.topic": kafka_topic,
    "cps.subscription": cps_subscription,
    "cps.project": cps_project,
    "value.converter": value_converter,
    "key.converter": key_converter,
}

connector = Connector()
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

コネクタを作成したら、コネクタの編集、削除、一時停止、停止、再起動を行うことができます。

コネクタを構成する

このセクションでは、コネクタに設定できる構成プロパティについて説明します。

このコネクタに固有のプロパティの一覧については、Pub/Sub ソースコネクタの構成をご覧ください。

pull モード

pull モードでは、コネクタが Pub/Sub メッセージを取得する方法を指定します。次のモードがサポートされています。

• pull モード（デフォルト）。メッセージはバッチで pull されます。このモードを有効にするには、cps.streamingPull.enabled=false. を設定します。バッチサイズを構成するには、cps.maxBatchSize プロパティを設定します。

  pull モードの詳細については、Pull API をご覧ください。

• ストリーミング pull モード。Pub/Sub からメッセージを取得する際に、スループットを最大化し、レイテンシを最小化します。このモードを有効にするには、cps.streamingPull.enabled=true を設定します。

  ストリーミング pull モードの詳細については、StreamingPull API をご覧ください。

  ストリーミング pull が有効になっている場合は、次の構成プロパティを設定してパフォーマンスを調整できます。

  • cps.streamingPull.flowControlBytes: タスクあたりの未処理メッセージの最大バイト数。
  • cps.streamingPull.flowControlMessages: タスクあたりの未処理メッセージの最大数。
  • cps.streamingPull.maxAckExtensionMs: コネクタがサブスクライブの期限を延長する最大時間（ミリ秒単位）。
  • cps.streamingPull.maxMsPerAckExtension: コネクタが拡張機能ごとにサブスクライブの期限を延長する最大時間（ミリ秒単位）。
  • cps.streamingPull.parallelStreams: サブスクリプションからメッセージを pull するストリームの数。

Pub/Sub エンドポイント

デフォルトでは、コネクタはグローバル Pub/Sub エンドポイントを使用します。エンドポイントを指定するには、cps.endpoint プロパティをエンドポイント アドレスに設定します。エンドポイントの詳細については、Pub/Sub エンドポイントをご覧ください。

Kafka パーティション

デフォルトでは、コネクタはトピック内の単一のパーティションに書き込みます。コネクタが書き込むパーティションの数を指定するには、kafka.partition.count プロパティを設定します。値は、トピックのパーティション数以下にする必要があります。

コネクタがメッセージをパーティションに割り当てる方法を指定するには、kafka.partition.scheme プロパティを設定します。詳細については、Pub/Sub Source コネクタの構成をご覧ください。

コンバージョンに至ったユーザー

キーコンバータを org.apache.kafka.connect.storage.StringConverter に設定します。

コネクタの構成に応じて、値コンバータを次のいずれかに設定します。

• org.apache.kafka.connect.converters.ByteArrayConverter
• org.apache.kafka.connect.json.JsonConverter

詳細については、レコード値をご覧ください。

メッセージ コンバージョン

Pub/Sub ソースコネクタは、Pub/Sub メッセージを Kafka レコードに変換します。以降のセクションでは、変換プロセスについて説明します。

レコードキー

キーコンバータは org.apache.kafka.connect.storage.StringConverter にする必要があります。

• デフォルトでは、レコードキーは null です。

• Pub/Sub メッセージ属性をキーとして使用するには、kafka.key.attribute を属性の名前に設定します。例: kafka.key.attribute=username

• Pub/Sub の順序指定キーをキーとして使用するには、kafka.key.attribute=orderingKey を設定します。

レコード ヘッダー

デフォルトでは、レコードヘッダーは空です。

kafka.record.headers が true の場合、Pub/Sub メッセージ属性はレコード ヘッダーとして書き込まれます。順序指定キーを含めるには、cps.makeOrderingKeyAttribute=true を設定します。

レコード値

レコード値は、バイト配列または struct 型として書き込まれます。

バイト配列レコード値

kafka.record.headers が true であるか、Pub/Sub メッセージにカスタム属性がない場合、コネクタはメッセージ データをバイト配列として書き込みます。値コンバータを org.apache.kafka.connect.converters.ByteArrayConverter に設定します。

構造体レコード値

kafka.record.headers が false で、メッセージに 1 つ以上のカスタム属性がある場合、コネクタはレコード値を struct として書き込みます。値コンバータを org.apache.kafka.connect.json.JsonConverter に設定します。

struct には次のフィールドがあります。

• message: Pub/Sub メッセージ データ（バイト単位）。

• Pub/Sub メッセージ属性ごとに 1 つのフィールド。順序付けキーを含めるには、cps.makeOrderingKeyAttribute=true を設定します。

たとえば、メッセージに username 属性がある場合、レコード値は次のようになります。

{
  "message":"MESSAGE_DATA",
  "username":"Alice"
}

value.converter.schemas.enable が true の場合、struct にはペイロードとスキーマの両方が含まれます。

{
  "schema":
    {
      "type":"struct",
      "fields": [
        {
          "type":"bytes",
          "optional":false,
          "field":"message"
        },
        {
          "type":"string",
          "optional":false,
          "field":"username"
        }
      ],
      "optional":false
    },
    "payload": {
      "message":"MESSAGE_DATA",
      "username":"Alice"
    }
}

次のステップ

• Pub/Sub メッセージを Kafka に複製する

• Pub/Sub コネクタのトラブルシューティング

• コネクタの一時停止、再開、停止、再起動

• すべてのコネクタを一覧表示する

• コネクタの詳細を表示する

• コネクタを更新する

Apache Kafka® は、Apache Software Foundation または米国その他の諸国における関連会社の商標です。