メッセージエラーの処理

サブスクライバーは、さまざまな理由でメッセージを処理できない場合があります。たとえば、メッセージの処理に必要なデータを取得する際に一時的な問題が発生している可能性があります。サブスクライバーが想定していない形式のメッセージが含まれている場合もあります。

このページでは、サブスクリプションの再試行ポリシーを使用するか、未配信のメッセージをデッドレタートピック（デッドレターキュー）に転送することで、処理障害を処理する方法について説明します。

これらの機能は Dataflow ではサポートされていません。詳細については、Dataflow ドキュメントのサポートされていない Pub/Sub 機能をご覧ください。

サブスクリプションの再試行ポリシー

Pub/Sub がメッセージの配信を試みてもサブスクライバーによるメッセージの確認応答ができない場合、Pub/Sub はメッセージを自動的に再送信しようとします。この再配信の試みは、サブスクリプションの再試行ポリシーと呼ばれます。これは、オンまたはオフにできる機能ではありません。ただし、使用する再試行ポリシーのタイプは選択できます。

サブスクリプションを最初に作成して構成するときに、2 つの再試行ポリシー（即時再配信または指数バックオフ）のいずれかを選択できます。デフォルトでは、即時再配信が使用されます。

即時再配信

デフォルトでは、Pub/Sub はメッセージの再送信（同じサブスクライバークライアントへの場合もあります）を直ちに試行します。ただし、メッセージの確認応答を防止する条件が変更されていない場合、すぐに再配信を行うと問題が発生する可能性があります。この場合、確認応答できない複数のメッセージを Pub/Sub が再送信する可能性があります。

即時再配信の問題に対処するため、Pub/Sub では指数バックオフポリシーを構成できます。

指数バックオフ

指数バックオフを使用すると、再試行の間の遅延を徐々に長くすることができます。最初の配信が失敗すると、Pub/Sub は最小バックオフ時間だけ待機してから再試行します。メッセージが連続して失敗するたびに、最大遅延時間（0～600 秒）に達するまで遅延時間が長くなります。

最大と最小の遅延間隔は固定されておらず、アプリケーションに対するローカル要因に基づいて構成する必要があります。

指数バックオフについては、次の点を考慮してください。

次のアクションによって指数バックオフがトリガーされます。
- 否定確認応答を受け取った場合。
- メッセージの確認応答期限が切れた場合。
指数バックオフは、サブスクリプション内のすべてのメッセージ（グローバル）ではなく、メッセージ単位でのみ適用されます。
指数バックオフの使用中は、以前のメッセージが否定確認応答を受信しても、Pub/Sub は引き続き他のメッセージを配信します（順序付きメッセージ配信を使用している場合を除きます）。

再試行ポリシーを使用して、メッセージのサブセットの配信と処理を遅らせ、配信時に一部のメッセージを一時的に処理できないことに対処します。この機能はベストエフォート方式で適用され、各メッセージが個々の再試行ポリシーに対して評価されます。

この機能を使用して、メッセージの配信を意図的に遅らせることはおすすめしません。再試行ポリシーで構成されているサブスクリプションで多数のメッセージを否定応答した場合、それらのメッセージの一部は、少ないバックオフまたはバックオフなしで配信される可能性があります。また、多数のメッセージを否定応答した場合、Pub/Sub によってすべてのメッセージの配信速度が遅くなることがあります。

配信をスケジュールする必要がある場合は、Cloud Tasks の使用を検討してください。

指数バックオフを構成する

Console

新しいサブスクリプションを作成する際に、次の手順に沿って指数バックオフ再試行ポリシーを構成できます。

Google Cloud コンソールで、[Pub/Sub サブスクリプション] ページに移動します。

サブスクリプションに移動
[サブスクリプションを作成] をクリックします。
[サブスクリプション ID] フィールドに名前を入力します。

サブスクリプションの指定方法については、トピックまたはサブスクリプションの指定方法のガイドラインをご覧ください。
プルダウンメニューからトピックを選択するか、作成します。

サブスクリプションがトピックからメッセージを受信します。
[配信タイプ] を選択します。
[再試行ポリシー] で、[指数バックオフ後の再試行] を選択します。
[最小バックオフ] と [最大バックオフ] に 0～600 秒の値を入力します。

デフォルト値は最小バックオフでは 10 秒、最大バックオフでは 600 秒です。
[作成] をクリックします。

gcloud

指数バックオフの再試行ポリシーを使用して新しいサブスクリプションを作成するには、以下に示すフラグを指定して gcloud pubsub create コマンドを実行します。

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --min-retry-delay=MIN_RETRY_DELAY \
  --max-retry-delay=MAX_RETRY_DELAY

デッドレタートピック

Pub/Sub サービスがメッセージの配信を試みても、サブスクライバーが確認応答できない場合、Pub/Sub は配信不能メッセージをデッドレタートピックに転送できます。

Pub/Sub でのデッドレタートピックの仕組み

デッドレタートピックは、トピックプロパティではなく、サブスクリプションプロパティになります。つまり、トピックの作成時ではなく、サブスクリプションの作成時にデッドレタートピックを設定します。

デッドレタートピックを作成する場合、次のサブスクリプションプロパティを設定できます。

配信試行回数の上限: Pub/Sub が特定のメッセージに対して行う配信の試行回数を示す数値。構成された配信試行回数内にサブスクライバークライアントがメッセージを確認応答できない場合、メッセージはデッドレタートピックに転送されます。
- デフォルト値 = 5
- 最大値 = 100
- 最小値 = 5
デッドレタートピックを含むプロジェクト: デッドレタートピックがサブスクリプションとは異なるプロジェクトにある場合、そのデッドレタートピックがあるプロジェクトを指定する必要があります。サブスクリプションが接続されているトピックとは異なるトピックに、デッドレタートピックを設定します。

配達の最大試行回数の算出方法

Pub/Sub は、デッドレタートピックが正しく構成され、正しい IAM 権限が含まれている場合にのみ、配信の試行回数をカウントします。

Pub/Sub では、ベストエフォート方式でメッセージが転送されるため、配信の最大試行回数には誤差が生じる場合があります。

特に、非アクティブなサブスクライバーを持つ pull サブスクリプションの場合、メッセージの配信試行回数はゼロにリセットされることもあります。その結果、構成された最大配信試行回数を超えて、サブスクライバークライアントにメッセージが配信される可能性があります。

デッドレタートピックを構成する

デッドレタートピックを構成するには、まずソーストピックにサブスクリプションが必要です。デッドレタートピックは、サブスクリプションの作成時に指定できます。また、デッドレタートピックが存在するように既存のサブスクリプションを更新することもできます。

サブスクリプションでデッドレタリングを有効にするワークフローは次のとおりです。

デッドレタートピックを作成します。このトピックはソーストピックとは別のものです。
ソーストピックのサブスクリプションにデッドレタートピックを設定します。
デッドレタートピックからメッセージが失われないようにするには、少なくとも 1 つのサブスクリプションをデッドレタートピックに接続します。セカンダリサブスクリプションは、デッドレタートピックからメッセージを受信します。
Pub/Sub サービスアカウントにパブリッシャーとサブスクライバーのロールを付与します。詳細については、転送権限を付与するをご覧ください。

新しいサブスクリプションにデッドレタートピックを設定する

Google Cloud コンソール、Google Cloud CLI、クライアントライブラリ、または Pub/Sub API を使用して、サブスクリプションを作成してデッドレタートピックを設定できます。

Console

サブスクリプションを作成してデッドレタートピックを設定するには、次の手順を実行します。

Google Cloud コンソールで、[サブスクリプション] ページに移動します。

サブスクリプションに移動
[サブスクリプションを作成] をクリックします。
[サブスクリプション ID] を入力します。
プルダウンメニューからトピックを選択するか、作成します。

サブスクリプションがトピックからメッセージを受信します。
[配信エラー]セクションで、[配信エラーを有効にする]を選択します。
デッドレタートピックをプルダウンメニューから選択または作成します。
選択したデッドレタートピックにサブスクリプションがない場合は、作成するよう求められます。
[配信の最大試行回数] フィールドで、5〜100 の整数を指定します。
[作成] をクリックします。
詳細パネルには、実行可能なアクションアイテムのリストが表示されます。いずれかの項目にエラーアイコンが表示されている場合は、操作項目をクリックして問題を解決します。

gcloud

サブスクリプションを作成してデッドレタートピックを設定するには、gcloud pubsub subscriptions create コマンドを使用します。

gcloud pubsub subscriptions create subscription-id \
  --topic=topic-id \
  --dead-letter-topic=dead-letter-topic-name \
  [--max-delivery-attempts=max-delivery-attempts] \
  [--dead-letter-topic-project=dead-letter-topic-project]

C++

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の C++ の設定手順を実施してください。詳細については、Pub/Sub C++ API リファレンスドキュメントをご覧ください。

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id,
   std::string const& dead_letter_topic_id,
   int dead_letter_delivery_attempts) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.mutable_dead_letter_policy()->set_dead_letter_topic(
      pubsub::Topic(project_id, dead_letter_topic_id).FullName());
  request.mutable_dead_letter_policy()->set_max_delivery_attempts(
      dead_letter_delivery_attempts);
  auto sub = client.CreateSubscription(request);
  if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The subscription already exists\n";
    return;
  }
  if (!sub) throw std::move(sub).status();

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

  std::cout << "It will forward dead letter messages to: "
            << sub->dead_letter_policy().dead_letter_topic() << "\n";

  std::cout << "After " << sub->dead_letter_policy().max_delivery_attempts()
            << " delivery attempts.\n";
}

C#

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の C# の設定手順を実施してください。詳細については、Pub/Sub C# API リファレンスドキュメントをご覧ください。


using Google.Cloud.PubSub.V1;
using System;

public class CreateSubscriptionWithDeadLetterPolicySample
{
    public Subscription CreateSubscriptionWithDeadLetterPolicy(string projectId, string topicId, string subscriptionId, string deadLetterTopicId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        // This is the subscription you want to create with a dead letter policy.
        var subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
        // This is an existing topic that you want to attach the subscription with dead letter policy to.
        var topicName = TopicName.FromProjectTopic(projectId, topicId);
        // This is an existing topic that the subscription with dead letter policy forwards dead letter messages to.
        var deadLetterTopic = TopicName.FromProjectTopic(projectId, deadLetterTopicId).ToString();
        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            DeadLetterPolicy = new DeadLetterPolicy
            {
                DeadLetterTopic = deadLetterTopic,
                // The maximum number of times that the service attempts to deliver a
                // message before forwarding it to the dead letter topic. Must be [5-100].
                MaxDeliveryAttempts = 10
            },
            AckDeadlineSeconds = 30
        };

        var subscription = subscriber.CreateSubscription(subscriptionRequest);
        Console.WriteLine("Created subscription: " + subscription.SubscriptionName.SubscriptionId);
        Console.WriteLine($"It will forward dead letter messages to: {subscription.DeadLetterPolicy.DeadLetterTopic}");
        Console.WriteLine($"After {subscription.DeadLetterPolicy.MaxDeliveryAttempts} delivery attempts.");
        // Remember to attach a subscription to the dead letter topic because
        // messages published to a topic with no subscriptions are lost.
        return subscription;
    }
}

Go

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の Go の設定手順を実施してください。詳細については、Pub/Sub Go API リファレンスドキュメントをご覧ください。

import (
	"context"
	"fmt"
	"io"
	"time"

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

// createSubWithDeadLetter creates a subscription with a dead letter policy.
func createSubWithDeadLetter(w io.Writer, projectID, subID string, topicID string, fullyQualifiedDeadLetterTopic string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topicID := "my-topic"
	// fullyQualifiedDeadLetterTopic := "projects/my-project/topics/my-dead-letter-topic"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	topic := client.Topic(topicID)

	subConfig := pubsub.SubscriptionConfig{
		Topic:       topic,
		AckDeadline: 20 * time.Second,
		DeadLetterPolicy: &pubsub.DeadLetterPolicy{
			DeadLetterTopic:     fullyQualifiedDeadLetterTopic,
			MaxDeliveryAttempts: 10,
		},
	}

	sub, err := client.CreateSubscription(ctx, subID, subConfig)
	if err != nil {
		return fmt.Errorf("CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created subscription (%s) with dead letter topic (%s)\n", sub.String(), fullyQualifiedDeadLetterTopic)
	fmt.Fprintln(w, "To process dead letter messages, remember to add a subscription to your dead letter topic.")
	return nil
}

Java

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の Java の設定手順を実施してください。詳細については、Pub/Sub Java API リファレンスドキュメントをご覧ください。

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.DeadLetterPolicy;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateSubscriptionWithDeadLetterPolicyExample {

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    // This is the subscription you want to create with a dead letter policy.
    String subscriptionId = "your-subscription-id";
    // This is an existing topic that you want to attach the subscription with dead letter policy
    // to.
    String topicId = "your-topic-id";
    // This is an existing topic that the subscription with dead letter policy forwards dead letter
    // messages to.
    String deadLetterTopicId = "your-dead-letter-topic-id";

    CreateSubscriptionWithDeadLetterPolicyExample.createSubscriptionWithDeadLetterPolicyExample(
        projectId, subscriptionId, topicId, deadLetterTopicId);
  }

  public static void createSubscriptionWithDeadLetterPolicyExample(
      String projectId, String subscriptionId, String topicId, String deadLetterTopicId)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);
      ProjectTopicName deadLetterTopicName = ProjectTopicName.of(projectId, deadLetterTopicId);

      DeadLetterPolicy deadLetterPolicy =
          DeadLetterPolicy.newBuilder()
              .setDeadLetterTopic(deadLetterTopicName.toString())
              // The maximum number of times that the service attempts to deliver a
              // message before forwarding it to the dead letter topic. Must be [5-100].
              .setMaxDeliveryAttempts(10)
              .build();

      Subscription request =
          Subscription.newBuilder()
              .setName(subscriptionName.toString())
              .setTopic(topicName.toString())
              .setDeadLetterPolicy(deadLetterPolicy)
              .build();

      Subscription subscription = subscriptionAdminClient.createSubscription(request);

      System.out.println("Created subscription: " + subscription.getName());
      System.out.println(
          "It will forward dead letter messages to: "
              + subscription.getDeadLetterPolicy().getDeadLetterTopic());
      System.out.println(
          "After "
              + subscription.getDeadLetterPolicy().getMaxDeliveryAttempts()
              + " delivery attempts.");
      // Remember to attach a subscription to the dead letter topic because
      // messages published to a topic with no subscriptions are lost.
    }
  }
}

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 subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const deadLetterTopicNameOrId = 'YOUR_DEAD_LETTER_TOPIC_NAME_OR_ID';

// 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 createSubscriptionWithDeadLetterPolicy(
  topicNameOrId,
  subscriptionNameOrId,
  deadLetterTopicNameOrId
) {
  // Creates a new subscription
  const options = {
    deadLetterPolicy: {
      deadLetterTopic: pubSubClient.topic(deadLetterTopicNameOrId).name,
      maxDeliveryAttempts: 10,
    },
  };
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, options);
  console.log(
    `Created subscription ${subscriptionNameOrId} with dead letter topic ${deadLetterTopicNameOrId}.`
  );
  console.log(
    'To process dead letter messages, remember to add a subscription to your dead letter topic.'
  );
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const deadLetterTopicNameOrId = 'YOUR_DEAD_LETTER_TOPIC_NAME_OR_ID';

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

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

async function createSubscriptionWithDeadLetterPolicy(
  topicNameOrId: string,
  subscriptionNameOrId: string,
  deadLetterTopicNameOrId: string
) {
  // Creates a new subscription
  const options: CreateSubscriptionOptions = {
    deadLetterPolicy: {
      deadLetterTopic: pubSubClient.topic(deadLetterTopicNameOrId).name,
      maxDeliveryAttempts: 10,
    },
  };
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, options);
  console.log(
    `Created subscription ${subscriptionNameOrId} with dead letter topic ${deadLetterTopicNameOrId}.`
  );
  console.log(
    'To process dead letter messages, remember to add a subscription to your dead letter topic.'
  );
}

PHP

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の PHP の設定手順を実施してください。詳細については、Pub/Sub PHP API リファレンスドキュメントをご覧ください。

use Google\Cloud\PubSub\PubSubClient;

/**
 * Creates a Pub/Sub subscription with dead letter policy enabled.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 * @param string $deadLetterTopicName The Pub/Sub topic to use for dead letter policy.
 */
function dead_letter_create_subscription($projectId, $topicName, $subscriptionName, $deadLetterTopicName)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);

    $topic = $pubsub->topic($topicName);
    $deadLetterTopic = $pubsub->topic($deadLetterTopicName);

    $subscription = $topic->subscribe($subscriptionName, [
        'deadLetterPolicy' => [
            'deadLetterTopic' => $deadLetterTopic
        ]
    ]);

    printf(
        'Subscription %s created with dead letter topic %s' . PHP_EOL,
        $subscription->name(),
        $deadLetterTopic->name()
    );
}

Python

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の Python の設定手順を実施してください。詳細については、Pub/Sub Python API リファレンスドキュメントをご覧ください。

from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import DeadLetterPolicy

# TODO(developer)
# project_id = "your-project-id"
# endpoint = "https://my-test-project.appspot.com/push"
# TODO(developer): This is an existing topic that the subscription
# with dead letter policy is attached to.
# topic_id = "your-topic-id"
# TODO(developer): This is an existing subscription with a dead letter policy.
# subscription_id = "your-subscription-id"
# TODO(developer): This is an existing dead letter topic that the subscription
# with dead letter policy will forward dead letter messages to.
# dead_letter_topic_id = "your-dead-letter-topic-id"
# TODO(developer): This is the maximum number of delivery attempts allowed
# for a message before it gets delivered to a dead letter topic.
# max_delivery_attempts = 5

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)
dead_letter_topic_path = publisher.topic_path(project_id, dead_letter_topic_id)

dead_letter_policy = DeadLetterPolicy(
    dead_letter_topic=dead_letter_topic_path,
    max_delivery_attempts=max_delivery_attempts,
)

with subscriber:
    request = {
        "name": subscription_path,
        "topic": topic_path,
        "dead_letter_policy": dead_letter_policy,
    }
    subscription = subscriber.create_subscription(request)

print(f"Subscription created: {subscription.name}")
print(
    f"It will forward dead letter messages to: {subscription.dead_letter_policy.dead_letter_topic}."
)
print(
    f"After {subscription.dead_letter_policy.max_delivery_attempts} delivery attempts."
)

Ruby

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の Ruby の設定手順を実施してください。詳細については、Pub/Sub Ruby API のリファレンスドキュメントをご覧ください。

Ruby

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Ruby 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Ruby API のリファレンスドキュメントをご覧ください。

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

# topic_id             = "your-topic-id"
# subscription_id      = "your-subscription-id"
# dead_letter_topic_id = "your-dead-letter-topic-id"

pubsub = Google::Cloud::Pubsub.new

topic             = pubsub.topic topic_id
dead_letter_topic = pubsub.topic dead_letter_topic_id
subscription      = topic.subscribe subscription_id,
                                    dead_letter_topic:                 dead_letter_topic,
                                    dead_letter_max_delivery_attempts: 10

puts "Created subscription #{subscription_id} with dead letter topic #{dead_letter_topic_id}."
puts "To process dead letter messages, remember to add a subscription to your dead letter topic."

既存のサブスクリプションのデッドレタートピックを設定する

Google Cloud コンソール、Google Cloud CLI、クライアントライブラリ、または Pub/Sub API を使用して、サブスクリプションを更新してデッドレタートピックを設定できます。

Console

サブスクリプションを更新してデッドレタートピックを設定するには、次の手順を実行します。

Google Cloud コンソールで、[サブスクリプション] ページに移動します。

サブスクリプションに移動
更新するサブスクリプションの横にある [その他の操作] アイコンをクリックします。
コンテキストメニューで [編集] を選択します。
[配信エラー]セクションで、[配信エラーを有効にする]を選択します。
プルダウンメニューからトピックを選択するか、作成します。
選択したトピックにサブスクリプションがない場合は、サブスクリプションを作成するよう求められます。
[配信の最大試行回数] フィールドで、5〜100 の整数を指定します。
[更新] をクリックします。
詳細パネルには、実行可能なアクションアイテムのリストが表示されます。いずれかの項目にエラーアイコンが表示されている場合は、操作項目をクリックして問題を解決します。

gcloud

サブスクリプションを更新してデッドレタートピックを設定するには、gcloud pubsub subscriptions update コマンドを使用します。

gcloud pubsub subscriptions update subscription-id \
  --dead-letter-topic=dead-letter-topic-name \
  [--max-delivery-attempts=max-delivery-attempts] \
  [--dead-letter-topic-project=dead-letter-topic-project]

C++

namespace pubsub_admin = ::google::cloud::pubsub_admin;
namespace pubsub = ::google::cloud::pubsub;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& subscription_id,
   std::string const& dead_letter_topic_id,
   int dead_letter_delivery_attempts) {
  google::pubsub::v1::UpdateSubscriptionRequest request;
  request.mutable_subscription()->set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.mutable_subscription()
      ->mutable_dead_letter_policy()
      ->set_dead_letter_topic(
          pubsub::Topic(project_id, dead_letter_topic_id).FullName());
  request.mutable_subscription()
      ->mutable_dead_letter_policy()
      ->set_max_delivery_attempts(dead_letter_delivery_attempts);
  *request.mutable_update_mask()->add_paths() = "dead_letter_policy";
  auto sub = client.UpdateSubscription(request);
  if (!sub) throw std::move(sub).status();

  std::cout << "The subscription has been updated to: " << sub->DebugString()
            << "\n";

  std::cout << "It will forward dead letter messages to: "
            << sub->dead_letter_policy().dead_letter_topic() << "\n";

  std::cout << "After " << sub->dead_letter_policy().max_delivery_attempts()
            << " delivery attempts.\n";
}

C#


using Google.Cloud.PubSub.V1;
using Google.Protobuf.WellKnownTypes;

public class UpdateDeadLetterPolicySample
{
    public Subscription UpdateDeadLetterPolicy(string projectId, string topicId, string subscriptionId, string deadLetterTopicId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        // This is an existing topic that the subscription with dead letter policy is attached to.
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        // This is an existing subscription with a dead letter policy.
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
        // This is an existing dead letter topic that the subscription with dead letter policy forwards
        // dead letter messages to.
        var deadLetterTopic = TopicName.FromProjectTopic(projectId, deadLetterTopicId).ToString();


        // Construct the subscription with the dead letter policy you expect to have after the update.
        // Here, values in the required fields (name, topic) help identify the subscription.
        var subscription = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            DeadLetterPolicy = new DeadLetterPolicy
            {
                DeadLetterTopic = deadLetterTopic,
                MaxDeliveryAttempts = 20,
            }
        };

        var request = new UpdateSubscriptionRequest
        {
            Subscription = subscription,
            // Construct a field mask to indicate which field to update in the subscription.
            UpdateMask = new FieldMask { Paths = { "dead_letter_policy" } }
        };
        var updatedSubscription = subscriber.UpdateSubscription(request);
        return updatedSubscription;
    }
}

Go

import (
	"context"
	"fmt"
	"io"

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

// updateDeadLetter updates an existing subscription with a dead letter policy.
func updateDeadLetter(w io.Writer, projectID, subID string, fullyQualifiedDeadLetterTopic string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// fullyQualifiedDeadLetterTopic := "projects/my-project/topics/my-dead-letter-topic"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	updateConfig := pubsub.SubscriptionConfigToUpdate{
		DeadLetterPolicy: &pubsub.DeadLetterPolicy{
			DeadLetterTopic:     fullyQualifiedDeadLetterTopic,
			MaxDeliveryAttempts: 20,
		},
	}

	subConfig, err := client.Subscription(subID).Update(ctx, updateConfig)
	if err != nil {
		return fmt.Errorf("Update: %w", err)
	}
	fmt.Fprintf(w, "Updated subscription config: %+v\n", subConfig)
	return nil
}

Java

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の Java の設定手順を実施してください。詳細については、Pub/Sub Java API のリファレンスドキュメントをご覧ください。


import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.protobuf.FieldMask;
import com.google.pubsub.v1.DeadLetterPolicy;
import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.SubscriptionName;
import com.google.pubsub.v1.TopicName;
import com.google.pubsub.v1.UpdateSubscriptionRequest;
import java.io.IOException;

public class UpdateDeadLetterPolicyExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    // This is an existing subscription with a dead letter policy.
    String subscriptionId = "your-subscription-id";
    // This is an existing topic that the subscription with dead letter policy is attached to.
    String topicId = "your-topic-id";
    // This is an existing dead letter topic that the subscription with dead letter policy forwards
    // dead letter messages to.
    String deadLetterTopicId = "your-dead-letter-topic-id";

    UpdateDeadLetterPolicyExample.updateDeadLetterPolicyExample(
        projectId, subscriptionId, topicId, deadLetterTopicId);
  }

  public static void updateDeadLetterPolicyExample(
      String projectId, String subscriptionId, String topicId, String deadLetterTopicId)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);

      System.out.println(
          "Before: " + subscriptionAdminClient.getSubscription(subscriptionName).getAllFields());

      TopicName topicName = TopicName.of(projectId, topicId);
      TopicName deadLetterTopicName = TopicName.of(projectId, deadLetterTopicId);

      // Construct the dead letter policy you expect to have after the update.
      DeadLetterPolicy deadLetterPolicy =
          DeadLetterPolicy.newBuilder()
              .setDeadLetterTopic(deadLetterTopicName.toString())
              .setMaxDeliveryAttempts(20)
              .build();

      // Construct the subscription with the dead letter policy you expect to have
      // after the update. Here, values in the required fields (name, topic) help
      // identify the subscription.
      Subscription subscription =
          Subscription.newBuilder()
              .setName(subscriptionName.toString())
              .setTopic(topicName.toString())
              .setDeadLetterPolicy(deadLetterPolicy)
              .build();

      // Construct a field mask to indicate which field to update in the subscription.
      FieldMask updateMask =
          FieldMask.newBuilder().addPaths("dead_letter_policy.max_delivery_attempts").build();

      UpdateSubscriptionRequest request =
          UpdateSubscriptionRequest.newBuilder()
              .setSubscription(subscription)
              .setUpdateMask(updateMask)
              .build();

      Subscription response = subscriptionAdminClient.updateSubscription(request);

      System.out.println("After: " + response.getAllFields());
      System.out.println(
          "Max delivery attempts is now "
              + response.getDeadLetterPolicy().getMaxDeliveryAttempts());
    }
  }
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// 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 updateDeadLetterPolicy(topicNameOrId, subscriptionNameOrId) {
  const metadata = {
    deadLetterPolicy: {
      deadLetterTopic: pubSubClient.topic(topicNameOrId).name,
      maxDeliveryAttempts: 15,
    },
  };

  await pubSubClient
    .topic(topicNameOrId)
    .subscription(subscriptionNameOrId)
    .setMetadata(metadata);

  console.log('Max delivery attempts updated successfully.');
}

PHP

use Google\Cloud\PubSub\PubSubClient;

/**
 * Set the dead letter policy on an existing subscription.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $deadLetterTopicName The Pub/Sub topic to use for dead letter policy.
 */
function dead_letter_update_subscription(
    string $projectId,
    string $topicName,
    string $subscriptionName,
    string $deadLetterTopicName
): void {
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);

    $topic = $pubsub->topic($topicName);
    $deadLetterTopic = $pubsub->topic($deadLetterTopicName);

    $subscription = $topic->subscription($subscriptionName);
    $subscription->update([
        'deadLetterPolicy' => [
            'deadLetterTopic' => $deadLetterTopic
        ]
    ]);

    printf(
        'Subscription %s updated with dead letter topic %s' . PHP_EOL,
        $subscription->name(),
        $deadLetterTopic->name()
    );
}

Python

from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import DeadLetterPolicy, FieldMask

# TODO(developer)
# project_id = "your-project-id"
# TODO(developer): This is an existing topic that the subscription
# with dead letter policy is attached to.
# topic_id = "your-topic-id"
# TODO(developer): This is an existing subscription with a dead letter policy.
# subscription_id = "your-subscription-id"
# TODO(developer): This is an existing dead letter topic that the subscription
# with dead letter policy will forward dead letter messages to.
# dead_letter_topic_id = "your-dead-letter-topic-id"
# TODO(developer): This is the maximum number of delivery attempts allowed
# for a message before it gets delivered to a dead letter topic.
# max_delivery_attempts = 5

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)
dead_letter_topic_path = publisher.topic_path(project_id, dead_letter_topic_id)

subscription_before_update = subscriber.get_subscription(
    request={"subscription": subscription_path}
)
print(f"Before the update: {subscription_before_update}.")

# Indicates which fields in the provided subscription to update.
update_mask = FieldMask(paths=["dead_letter_policy"])

# Construct a dead letter policy you expect to have after the update.
dead_letter_policy = DeadLetterPolicy(
    dead_letter_topic=dead_letter_topic_path,
    max_delivery_attempts=max_delivery_attempts,
)

# Construct the subscription with the dead letter policy you expect to have
# after the update. Here, values in the required fields (name, topic) help
# identify the subscription.
subscription = pubsub_v1.types.Subscription(
    name=subscription_path,
    topic=topic_path,
    dead_letter_policy=dead_letter_policy,
)

with subscriber:
    subscription_after_update: gapic_types.Subscription = (
        subscriber.update_subscription(
            request={"subscription": subscription, "update_mask": update_mask}
        )
    )

print(f"After the update: {subscription_after_update}.")

Ruby

# subscription_id       = "your-subscription-id"
# role                  = "roles/pubsub.publisher"
# service_account_email = "serviceAccount:account_name@project_name.iam.gserviceaccount.com"

pubsub = Google::Cloud::Pubsub.new

subscription = pubsub.subscription subscription_id
subscription.dead_letter_max_delivery_attempts = 20
puts "Max delivery attempts is now #{subscription.dead_letter_max_delivery_attempts}."

デッドレタートピックを使用するための IAM ロールを付与する

配信不能メッセージをデッドレタートピックに転送するには、Pub/Sub に次の権限が付与されている必要があります。

トピックにメッセージをパブリッシュする。
メッセージに確認応答することにより、メッセージをサブスクリプションから削除します。

Pub/Sub は、プロジェクト service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com ごとにサービスアカウントを作成して維持します。このサービスアカウントにパブリッシャーとサブスクライバーのロールを割り当てることで、転送権限を付与できます。

Console

デッドレタートピックにメッセージをパブリッシュする権限を Pub/Sub に付与するには、次の手順を行います。

Google Cloud コンソールで、[サブスクリプション] ページに移動します。

サブスクリプションに移動
デッドレタートピックを含むサブスクリプションの名前をクリックします。
[デッドレタリング] タブをクリックします。
パブリッシャーのロールを割り当てるには、[パブリッシャーのロールを付与する] をクリックします。パブリッシャーロールが正常に割り当てられていると、青いチェックマークが表示されます。
サブスクライバーのロールを割り当てるには、[サブスクライバーのロールを付与する] をクリックします。パブリッシャーロールが正常に割り当てられていると、青いチェックマークが表示されます。

gcloud

デッドレタートピックにメッセージをパブリッシュする権限を Pub/Sub に付与するには、次のコマンドを実行します。

PUBSUB_SERVICE_ACCOUNT="service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com"

gcloud pubsub topics add-iam-policy-binding dead-letter-topic-name \
  --member="serviceAccount:$PUBSUB_SERVICE_ACCOUNT"\
  --role="roles/pubsub.publisher"

転送された配信不能メッセージに確認応答する権限を Pub/Sub に付与するには、次のコマンドを実行します。

PUBSUB_SERVICE_ACCOUNT="service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com"

gcloud pubsub subscriptions add-iam-policy-binding subscription-id \
  --member="serviceAccount:$PUBSUB_SERVICE_ACCOUNT"\
  --role="roles/pubsub.subscriber"

配信試行の確認

サブスクリプションのデッドレタートピックを有効にすると、そのサブスクリプションのすべてのメッセージに配信試行回数を指定するフィールドが表示されます。

pull サブスクリプションから受信したメッセージには、delivery_attempt フィールドが含まれます。

注: Java または C# クライアントライブラリを使用する場合、配信試行は delivery_attempt フィールドではなく、googclient_deliveryattempt メッセージ属性にあります。
push サブスクリプションから受信したメッセージには、deliveryAttempt フィールドが含まれます。

次のサンプルは、配信の試行回数を取得する方法を示しています。

C++

このサンプルを試す前に、クイックスタート: クライアントライブラリの使用の C++ の設定手順を実施してください。詳細については、Pub/Sub C++ API のリファレンスドキュメントをご覧ください。

namespace pubsub = ::google::cloud::pubsub;
auto sample = [](pubsub::Subscriber subscriber) {
  return subscriber.Subscribe(
      [&](pubsub::Message const& m, pubsub::AckHandler h) {
        std::cout << "Received message " << m << "\n";
        std::cout << "Delivery attempt: " << h.delivery_attempt() << "\n";
        std::move(h).ack();
        PleaseIgnoreThisSimplifiesTestingTheSamples();
      });
};

C#

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある C# 向けの手順に従って設定を行ってください。詳細については、Pub/Sub C# API のリファレンスドキュメントをご覧ください。


using Google.Cloud.PubSub.V1;
using System.Threading;
using System.Threading.Tasks;

public class PullMessagesAsyncWithDeliveryAttemptsSample
{
    public async Task<int> PullMessagesAsyncWithDeliveryAttempts(string projectId, string subscriptionId, bool acknowledge)
    {
        // This is an existing subscription with a dead letter policy.
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        SubscriberClient subscriber = await SubscriberClient.CreateAsync(subscriptionName);

        int deliveryAttempt = 0;
        Task startTask = subscriber.StartAsync((PubsubMessage message, CancellationToken cancel) =>
        {
            string text = message.Data.ToStringUtf8();
            System.Console.WriteLine($"Delivery Attempt: {message.GetDeliveryAttempt()}");
            if (message.GetDeliveryAttempt() != null)
            {
                deliveryAttempt = message.GetDeliveryAttempt().Value;
            }
            return Task.FromResult(acknowledge ? SubscriberClient.Reply.Ack : SubscriberClient.Reply.Nack);
        });
        // Run for 7 seconds.
        await Task.Delay(7000);
        await subscriber.StopAsync(CancellationToken.None);
        // Lets make sure that the start task finished successfully after the call to stop.
        await startTask;
        return deliveryAttempt;
    }
}

Go

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Go 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Go API のリファレンスドキュメントをご覧ください。

import (
	"context"
	"fmt"
	"io"
	"time"

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

func pullMsgsDeadLetterDeliveryAttempt(w io.Writer, projectID, subID string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	// Receive messages for 10 seconds, which simplifies testing.
	// Comment this out in production, since `Receive` should
	// be used as a long running operation.
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	sub := client.Subscription(subID)
	err = sub.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
		// When dead lettering is enabled, the delivery attempt field is a pointer to the
		// the number of times the service has attempted to delivery a message.
		// Otherwise, the field is nil.
		if msg.DeliveryAttempt != nil {
			fmt.Fprintf(w, "message: %s, delivery attempts: %d", msg.Data, *msg.DeliveryAttempt)
		}
		msg.Ack()
	})
	if err != nil {
		return fmt.Errorf("got error in Receive: %w", err)
	}
	return nil
}

Java

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Java 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Java API のリファレンスドキュメントをご覧ください。


import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ReceiveMessagesWithDeliveryAttemptsExample {

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    // This is an existing subscription with a dead letter policy.
    String subscriptionId = "your-subscription-id";

    ReceiveMessagesWithDeliveryAttemptsExample.receiveMessagesWithDeliveryAttemptsExample(
        projectId, subscriptionId);
  }

  public static void receiveMessagesWithDeliveryAttemptsExample(
      String projectId, String subscriptionId) {

    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        new MessageReceiver() {
          @Override
          public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
            // Handle incoming message, then ack the received message.
            System.out.println("Id: " + message.getMessageId());
            System.out.println("Data: " + message.getData().toStringUtf8());
            System.out.println("Delivery Attempt: " + Subscriber.getDeliveryAttempt(message));
            consumer.ack();
          }
        };

    Subscriber subscriber = null;
    try {
      subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
      // Start the subscriber.
      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (TimeoutException timeoutException) {
      // Shut down the subscriber after 30s. Stop receiving messages.
      subscriber.stopAsync();
    }
  }
}

Node.js

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Node.js 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Node.js API のリファレンスドキュメントをご覧ください。

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const projectId = 'YOUR_PROJECT_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// Imports the Google Cloud client library. v1 is for the lower level
// proto access.
const {v1} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use.
const subClient = new v1.SubscriberClient();

async function synchronousPullWithDeliveryAttempts(
  projectId,
  subscriptionNameOrId
) {
  // The low level API client requires a name only.
  const formattedSubscription =
    subscriptionNameOrId.indexOf('/') >= 0
      ? subscriptionNameOrId
      : subClient.subscriptionPath(projectId, subscriptionNameOrId);

  // The maximum number of messages returned for this request.
  // Pub/Sub may return fewer than the number specified.
  const request = {
    subscription: formattedSubscription,
    maxMessages: 10,
  };

  // The subscriber pulls a specified number of messages.
  const [response] = await subClient.pull(request);

  // Process the messages.
  const ackIds = [];
  for (const message of response.receivedMessages || []) {
    console.log(`Received message: ${message.message.data}`);
    console.log(`Delivery Attempt: ${message.deliveryAttempt}`);
    if (message.ackId) {
      ackIds.push(message.ackId);
    }
  }

  // Acknowledge all of the messages. You could also acknowledge
  // these individually, but this is more efficient.
  const ackRequest = {
    subscription: formattedSubscription,
    ackIds: ackIds,
  };
  await subClient.acknowledge(ackRequest);

  console.log('Done.');
}

PHP

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある PHP 向けの手順に従って設定を行ってください。詳細については、Pub/Sub PHP API のリファレンスドキュメントをご覧ください。

use Google\Cloud\PubSub\Message;
use Google\Cloud\PubSub\PubSubClient;

/**
 * Get the delivery attempt from a pulled message.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 * @param string $message The contents of a pubsub message data field.
 */
function dead_letter_delivery_attempt($projectId, $topicName, $subscriptionName, $message)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);

    $topic = $pubsub->topic($topicName);

    // publish test message
    $topic->publish(new Message([
        'data' => $message
    ]));

    $subscription = $topic->subscription($subscriptionName);
    $messages = $subscription->pull();

    foreach ($messages as $message) {
        printf('Received message %s' . PHP_EOL, $message->data());
        printf('Delivery attempt %d' . PHP_EOL, $message->deliveryAttempt());
    }
    print('Done' . PHP_EOL);
}

Python

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Python 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Python API のリファレンスドキュメントをご覧ください。

from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1

# TODO(developer)
# project_id = "your-project-id"
# subscription_id = "your-subscription-id"

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    print(f"Received {message}.")
    print(f"With delivery attempts: {message.delivery_attempt}.")
    message.ack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}..\n")

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    # When `timeout` is not set, result() will block indefinitely,
    # unless an exception is encountered first.
    try:
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.

Ruby

# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::Pubsub.new

subscription = pubsub.subscription subscription_id
subscription.pull(immediate: false).each do |message|
  puts "Received message: #{message.data}"
  puts "Delivery Attempt: #{message.delivery_attempt}"
  message.acknowledge!
end

Pub/Sub が配信不能メッセージをデッドレタートピックに転送すると、次の属性がメッセージに追加されます。

CloudPubSubDeadLetterSourceDeliveryCount: ソースサブスクリプションへの配信試行回数。
CloudPubSubDeadLetterSourceSubscription: ソースサブスクリプションの名前。
CloudPubSubDeadLetterSourceSubscriptionProject: ソースサブスクリプションを含むプロジェクトの名前。
CloudPubSubDeadLetterSourceTopicPublishTime: メッセージが最初にパブリッシュされたときのタイムスタンプ。
CloudPubSubDeadLetterSourceDeliveryErrorMessage: 元の宛先にメッセージを配信できなかった理由。この属性は、サブスクリプションのエクスポートでのみ使用できます。

転送されたメッセージのモニタリング

配信不能メッセージを転送すると、Pub/Sub サービスは、サブスクリプションからメッセージを削除します。転送されるメッセージは、Cloud Monitoring を使用してモニタリングできます。

サブスクリプションをデッドレタートピックに接続すると、メッセージには、デッドレタートピックプロパティの有効期限ではなく、接続したサブスクリプションの有効期限ポリシーが使用されます。

subscription/dead_letter_message_count 指標には、Pub/Sub が転送するサブスクリプションの配信不能メッセージの数が記録されます。

詳細については、転送された配信不能メッセージのモニタリングをご覧ください。

デッドレタートピックを削除する

配信不能メッセージの転送を停止するには、サブスクリプションからデッドレタートピックを削除します。

サブスクリプションからデッドレタートピックを削除するには、Google Cloud コンソール、Google Cloud CLI、または Pub/Sub API を使用できます。

Console

サブスクリプションからデッドレタートピックを削除するには、次の手順を行います。

Google Cloud コンソールで、[サブスクリプション] ページに移動します。

サブスクリプションに移動
サブスクリプションのリストで、更新するサブスクリプションの横にあるをクリックします。
コンテキストメニューから [編集] を選択します。
[配信エラー]セクションで、[配信エラーを有効にする]をオフにします。
[更新] をクリックします。

gcloud

サブスクリプションからデッドレタートピックを削除するには、--clear-dead-letter-policy フラグを付けた gcloud pubsub subscriptions update コマンドを使用します。

gcloud pubsub subscriptions update subscription-id \
  --clear-dead-letter-policy

C++

namespace pubsub_admin = ::google::cloud::pubsub_admin;
namespace pubsub = ::google::cloud::pubsub;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& subscription_id) {
  google::pubsub::v1::UpdateSubscriptionRequest request;
  request.mutable_subscription()->set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.mutable_subscription()->clear_dead_letter_policy();
  *request.mutable_update_mask()->add_paths() = "dead_letter_policy";
  auto sub = client.UpdateSubscription(request);
  if (!sub) throw std::move(sub).status();

  std::cout << "The subscription has been updated to: " << sub->DebugString()
            << "\n";
}

C#


using Google.Cloud.PubSub.V1;
using Google.Protobuf.WellKnownTypes;

public class RemoveDeadLetterPolicySample
{
    public Subscription RemoveDeadLetterPolicy(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        // This is an existing topic that the subscription with dead letter policy is attached to.
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        // This is an existing subscription with dead letter policy.
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscription = new Subscription()
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            DeadLetterPolicy = null
        };

        var request = new UpdateSubscriptionRequest
        {
            Subscription = subscription,
            UpdateMask = new FieldMask { Paths = { "dead_letter_policy" } }
        };
        var updatedSubscription = subscriber.UpdateSubscription(request);
        return updatedSubscription;
    }
}

Go

import (
	"context"
	"fmt"
	"io"

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

// removeDeadLetterTopic removes the dead letter policy from a subscription.
func removeDeadLetterTopic(w io.Writer, projectID, subID string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	subConfig, err := client.Subscription(subID).Update(ctx, pubsub.SubscriptionConfigToUpdate{
		DeadLetterPolicy: &pubsub.DeadLetterPolicy{},
	})
	if err != nil {
		return fmt.Errorf("Update: %w", err)
	}
	fmt.Fprintf(w, "Updated subscription config: %+v\n", subConfig)
	return nil
}

Java


import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.protobuf.FieldMask;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.TopicName;
import com.google.pubsub.v1.UpdateSubscriptionRequest;

public class RemoveDeadLetterPolicyExample {

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    // This is an existing subscription with dead letter policy.
    String subscriptionId = "your-subscription-id";
    // This is an existing topic that the subscription with dead letter policy is attached to.
    String topicId = "your-topic-id";

    RemoveDeadLetterPolicyExample.removeDeadLetterPolicyExample(projectId, subscriptionId, topicId);
  }

  public static void removeDeadLetterPolicyExample(
      String projectId, String subscriptionId, String topicId) throws Exception {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);
      TopicName topicName = TopicName.of(projectId, topicId);

      // Construct the subscription you expect to have after the request. Here,
      // values in the required fields (name, topic) help identify the subscription.
      // No dead letter policy is supplied.
      Subscription expectedSubscription =
          Subscription.newBuilder()
              .setName(subscriptionName.toString())
              .setTopic(topicName.toString())
              .build();

      // Construct a field mask to indicate which field to update in the subscription.
      FieldMask updateMask = FieldMask.newBuilder().addPaths("dead_letter_policy").build();

      UpdateSubscriptionRequest request =
          UpdateSubscriptionRequest.newBuilder()
              .setSubscription(expectedSubscription)
              .setUpdateMask(updateMask)
              .build();

      Subscription response = subscriptionAdminClient.updateSubscription(request);

      // You should see an empty dead letter topic field inside the dead letter policy.
      System.out.println("After: " + response.getAllFields());
    }
  }
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// 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 removeDeadLetterPolicy(topicNameOrId, subscriptionNameOrId) {
  const metadata = {
    deadLetterPolicy: null,
  };

  await pubSubClient
    .topic(topicNameOrId)
    .subscription(subscriptionNameOrId)
    .setMetadata(metadata);

  console.log(
    `Removed dead letter topic from ${subscriptionNameOrId} subscription.`
  );
}

PHP

use Google\Cloud\PubSub\PubSubClient;

/**
 * Remove dead letter policy from an existing subscription.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 */
function dead_letter_remove($projectId, $topicName, $subscriptionName)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);

    $topic = $pubsub->topic($topicName);

    $subscription = $topic->subscription($subscriptionName);

    // Provide deadLetterPolicy in the update mask, but omit from update fields to unset.
    $subscription->update([], [
        'updateMask' => [
            'deadLetterPolicy'
        ]
    ]);

    printf(
        'Removed dead letter topic from subscription %s' . PHP_EOL,
        $subscription->name()
    );
}

Python

from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FieldMask

# TODO(developer)
# project_id = "your-project-id"
# TODO(developer): This is an existing topic that the subscription
# with dead letter policy is attached to.
# topic_id = "your-topic-id"
# TODO(developer): This is an existing subscription with a dead letter policy.
# subscription_id = "your-subscription-id"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

subscription_before_update = subscriber.get_subscription(
    request={"subscription": subscription_path}
)
print(f"Before removing the policy: {subscription_before_update}.")

# Indicates which fields in the provided subscription to update.
update_mask = FieldMask(paths=["dead_letter_policy"])

# Construct the subscription (without any dead letter policy) that you
# expect to have after the update.
subscription = pubsub_v1.types.Subscription(
    name=subscription_path, topic=topic_path
)

with subscriber:
    subscription_after_update: gapic_types.Subscription = (
        subscriber.update_subscription(
            request={"subscription": subscription, "update_mask": update_mask}
        )
    )

print(f"After removing the policy: {subscription_after_update}.")

Ruby

# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::Pubsub.new

subscription = pubsub.subscription subscription_id
subscription.remove_dead_letter_policy
puts "Removed dead letter topic from #{subscription_id} subscription."

料金

Pub/Sub サービスが配信不能メッセージを転送すると、次の料金が適用されます。

デッドレタートピックを含むプロジェクトの請求先アカウントに対して、パブリッシュ手数料。
デッドレタートピックプロパティのあるサブスクリプションを含むプロジェクトに関連付けられた請求先アカウントに対して請求される送信メッセージの利用料金。

サブスクリプションのデッドレタートピックプロパティは設定しているが、デッドレタートピックのメッセージストレージロケーションポリシーを設定しておらず、トピックでサブスクリプションを含むリージョンが許可されていな場合は、送信メッセージのパブリッシュ料金も適用されます。

送信メッセージのパブリッシュ料金は、デッドレタートピックを含むプロジェクトに対して請求されます。詳細については、料金をご覧ください。

メッセージ エラーの処理

サブスクリプションの再試行ポリシー

即時再配信

指数バックオフ

指数バックオフを構成する

Console

gcloud

デッドレター トピック

Pub/Sub でのデッドレター トピックの仕組み

デッドレター トピックを構成する

新しいサブスクリプションにデッドレター トピックを設定する

Console

gcloud

C++

C#

Go

Java

Node.js

Node.js

PHP

Python

Ruby

Ruby

既存のサブスクリプションのデッドレター トピックを設定する

Console

gcloud

C++

C#

Go

Java

Node.js

PHP

Python

Ruby

Ruby

デッドレター トピックを使用するための IAM ロールを付与する

Console

gcloud

配信試行の確認

C++

C#

Go

Java

Node.js

PHP

Python

Ruby

転送されたメッセージのモニタリング

デッドレター トピックを削除する

Console

gcloud

C++

C#

Go

Java

Node.js

PHP

Python

Ruby

Ruby

料金

次のステップ

メッセージエラーの処理

デッドレタートピック

Pub/Sub でのデッドレタートピックの仕組み

デッドレタートピックを構成する

新しいサブスクリプションにデッドレタートピックを設定する

既存のサブスクリプションのデッドレタートピックを設定する

デッドレタートピックを使用するための IAM ロールを付与する

デッドレタートピックを削除する