Solución de problemas generales

Obtén información sobre los pasos para solucionar problemas que pueden servirte si tienes dificultades con Pub/Sub.

No se puede crear un tema

Verifica que tengas los permisos necesarios . Para crear un tema de Pub/Sub, necesitas el rol de Identity and Access Management de editor de Pub/Sub (roles/pubsub.editor) en el proyecto. Si no tienes este rol, comunícate con tu administrador. Para obtener más información sobre la solución de problemas relacionados con temas, consulta las siguientes páginas:

• Solución de problemas relacionados con temas

• Solución de problemas relacionados con temas de importación

• Solución de problemas relacionados con permisos

No se puede crear una suscripción

Verifica que hayas hecho lo siguiente:

• Verifica que tengas los permisos necesarios . Para crear una suscripción a Pub/Sub, necesitas el rol de IAM de editor de Pub/Sub (roles/pubsub.editor) en el proyecto. Si no tienes este rol, comunícate con tu administrador.

• Especificaste un nombre para la suscripción.

• Especificaste el nombre de un tema existente al que deseas adjuntar la suscripción.

• Si creas una suscripción de envío push, especificaste https:// en minúsculas (no http:// ni HTTPS://) como protocolo para tu URL de recepción en el campo pushEndpoint.

Para obtener más información sobre la solución de problemas relacionados con suscripciones, consulta las siguientes páginas:

• Solución de problemas relacionados con extracción, envío, BigQuery, o Cloud Storage

• Solución de problemas relacionados con suscripciones con transformaciones de mensajes individuales

• Solución de problemas relacionados con permisos

Soluciona problemas relacionados con permisos

Los permisos de Pub/Sub controlan qué usuarios y cuentas de servicio pueden realizar acciones en tus recursos de Pub/Sub. Cuando los permisos están mal configurados, pueden generar errores de permiso denegado y interrumpir el flujo de mensajes. Los registros de auditoría proporcionan un registro detallado de todos los cambios de permisos, lo que te permite identificar la fuente de estos problemas.

Para solucionar problemas de permisos de Pub/Sub con registros de auditoría, haz lo siguiente:

1. Obtén los permisos necesarios para ver el Explorador de registros.

   Para obtener más información, consulta Antes de comenzar.

2. En la Google Cloud consola de, accede a la página Explorador de registros.

   Ir al Explorador de registros

3. Elige una organización, un proyecto o una carpeta deexistentes Google Cloud .

4. Esta es una lista de filtros que puedes usar para encontrar registros relevantes:

   • resource.type="pubsub_topic" OR resource.type="pubsub_subscription":Usa esta consulta como punto de partida cuando solucionas problemas que puedan implicar cambios en las configuraciones de temas o suscripciones, o en el control de acceso. Puedes combinarla con otros filtros para refinar aún más tu búsqueda.

   • protoPayload.methodName="google.iam.v1.SetIamPolicy": Usa esta consulta cuando sospeches que un problema se debe a permisos incorrectos o faltantes Te ayuda a hacer un seguimiento de quién realizó cambios en la política de IAM y cuáles fueron esos cambios. Esto puede ser útil para solucionar problemas, como usuarios que no pueden publicar en temas o suscribirse a suscripciones, aplicaciones a las que se les denegó el acceso a recursos de Pub/Sub o cambios inesperados en el control de acceso.

   • protoPayload.status.code=7: Usa esta consulta cuando encuentres errores relacionados explícitamente con los permisos. Esto te ayuda a identificar qué acciones fallan y quién las intenta. Puedes combinar esta consulta con las anteriores para identificar el recurso específico y el cambio de política de IAM que podrían estar causando la denegación de permisos.

5. Analiza los registros para determinar factores como la marca de tiempo del evento, la entidad principal que realizó el cambio y el tipo de cambios que se realizaron.

6. Según la información recopilada de los registros de auditoría, puedes tomar medidas correctivas.

Soluciona problemas de permisos de Terraform

Cuando usas Pub/Sub con Terraform, otorga explícitamente los roles necesarios en tu código de Terraform. Por ejemplo, para la publicación, la cuenta de servicio de tu aplicación necesita el rol roles/pubsub.publisher. Si este rol no está definido de forma explícita en tu código de Terraform, un terraform apply futuro podría quitarlo. Esto suele ocurrir durante actualizaciones no relacionadas, lo que provoca que una aplicación confiable falle de repente con errores PERMISSION_DENIED. Definir el rol de forma explícita en tu código evita estas regresiones accidentales.

Se borró la suscripción

Las suscripciones a Pub/Sub se pueden borrar de dos maneras principales:

• Un usuario o una cuenta de servicio con permisos suficientes borra la suscripción de forma intencional.

• Una suscripción se borra automáticamente después de un período de inactividad, que es de 31 días de forma predeterminada. Para obtener más información sobre la política de vencimiento de la suscripción, consulta Período de vencimiento.

Para solucionar problemas relacionados con una suscripción borrada, sigue estos pasos:

1. En la Google Cloud consola de, ve a la página Suscripciones de Pub/Sub y verifica que la suscripción ya no aparezca en la lista. Para obtener más información sobre cómo enumerar suscripciones, consulta Enumera una suscripción.

2. Revisa los registros de auditoría. Navega al Explorador de registros. Usa el filtro protoPayload.methodName="google.pubsub.v1.Subscriber.DeleteSubscription" para encontrar suscripciones borradas. Examina los registros para determinar si alguien borró la suscripción o si se borró debido a la inactividad. InternalExpireInactiveSubscription indica que se borró una suscripción debido a la inactividad. Para obtener más información sobre cómo usar los registros de auditoría para solucionar problemas, consulta Soluciona problemas de Pub/Sub con registros de auditoría.

Error 403 (Forbidden)

Por lo general, un error 403 significa que no tienes los permisos correctos para realizar una acción. Por ejemplo, es posible que recibas un error 403 User not authorized cuando intentes publicar en un tema o extraer de una suscripción.

Si recibes este error, haz lo siguiente:

• Asegúrate de que habilitaste la API de Pub/Sub en la Google Cloud consola.

• Asegúrate de que la cuenta principal que realiza la solicitud tenga los permisos necesarios en los recursos relevantes de la API de Pub/Sub, en especial, si usas la API de Pub/Sub para la comunicación entre proyectos.

• Si usas Dataflow, asegúrate de que {PROJECT_NUMBER}@cloudservices.gserviceaccount.com y la cuenta de servicio de Compute Engine {PROJECT_NUMBER}[e-mail] tengan los permisos necesarios en el recurso de la API de Pub/Sub relevante. Para obtener más información, consulta Seguridad y permisos de Dataflow.

• Si usas App Engine, revisa la página de permisos de tu proyecto para ver si alguna cuenta de servicio de App Engine aparece como editor de Pub/Sub. Si no es así, agrega tu cuenta de servicio de App Engine como editor de Pub/Sub. Por lo general, la cuenta de servicio de App Engine tiene el formato <project-id>@appspot.gserviceaccount.com.

• Puedes usar registros de auditoría para solucionar problemas relacionados con permisos.

Otros códigos de error comunes

Para obtener una lista de otros códigos de error comunes relacionados con la API de Pub/Sub y sus descripciones, consulta Códigos de error.

Tiempos de espera agotados de conexión, latencia o errores de red

Es posible que experimentes fallas intermitentes o persistentes cuando tus aplicaciones cliente de Pub/Sub intenten conectarse a Google Cloud los servicios. Estos problemas pueden manifestarse de la siguiente manera:

• Demoras significativas cuando se publican mensajes, lo que puede provocar retrasos en las aplicaciones
• Errores de tiempo de espera agotado, como DEADLINE_EXCEEDED de gRPC, code = DeadlineExceeded o java.net.SocketTimeoutException
• Fallas de E/S de red, como UNAVAILABLE: io exception o errores Connection refused cuando se intenta acceder a servicios como pubsub.googleapis.com o oauth2.googleapis.com.

Estos problemas de conectividad pueden surgir incluso sin cambios en la configuración de Pub/Sub o en el código de la aplicación. Esto ocurre con frecuencia cuando los firewalls locales o de VPC usan listas de entidades permitidas de direcciones IP codificadas para las APIs de Google. Los servicios de Google, incluidos Pub/Sub y sus dependencias, como los servicios de autenticación, usan un rango dinámico de direcciones IP. Si tu firewall no tiene en cuenta las direcciones IP nuevas, puede bloquear el tráfico a las direcciones IP nuevas, lo que provoca fallas de conexión y autenticación.

Para garantizar una conectividad estable, evita las reglas de firewall basadas en IP estáticas para los servicios de Google. En cambio, puedes hacer lo siguiente:

• Configura tu firewall para permitir el tráfico con los rangos de IP publicados de Google para dominios predeterminados en lugar de direcciones codificadas. Para obtener información sobre cómo obtener estos rangos y automatizar las actualizaciones de tus reglas de firewall, consulta Direcciones IP para dominios predeterminados.
• Habilita el Acceso privado a Google, que permite que las instancias dentro de tu red de VPC accedan a las APIs de Google y a los servicios sin atravesar la Internet pública, lo que simplifica la administración del firewall.

JWT no válido: El token debe ser de corta duración

Si recibes un error como Invalid JWT: Token must be a short-lived token (60 minutes) and in a reasonable timeframe cuando tu aplicación interactúa con la API de Pub/Sub, esto suele indicar un problema con el tiempo de las credenciales de autenticación.

Este error se produce durante la validación del token web JSON (JWT) que se usa para autenticar las solicitudes a la API. Una causa común es una diferencia horaria significativa (desviación horaria) entre la máquina cliente que ejecuta la biblioteca de Pub/Sub y los servidores de autenticación de Google. Debido a que los JWT tienen un período de validez limitado, las discrepancias de reloj pueden hacer que se traten como vencidos o aún no válidos.

Para resolver este problema, sincroniza el reloj de tu máquina cliente:

• Verifica que la fecha, la hora y la zona horaria de tu máquina sean correctas.

• Usa un servicio de protocolo de hora de red (NTP) para mantener sincronizada la hora del sistema y verifica que el servicio se esté ejecutando y configurado correctamente.

Uso de operaciones administrativas excesivas

Si descubres que consumes demasiado tu cuota para operaciones administrativas, es posible que debas reestructurar el código. A modo de ilustración, considera este seudocódigo. En este ejemplo, se usa una operación administrativa (GET) para verificar la presencia de una suscripción antes de que intente consumir sus recursos. GET y CREATE son operaciones administrativas:

if !GetSubscription my-sub {
  CreateSubscription my-sub
}
Consume from subscription my-sub

Una opción más eficaz es intentar consumir los mensajes de la suscripción (siempre que puedas estar seguro del nombre de la suscripción). Con este enfoque optimista, solo obtienes o creas la suscripción si hay un error. Considera el siguiente ejemplo:

try {
  Consume from subscription my-sub
} catch NotFoundError {
  CreateSubscription my-sub
  Consume from subscription my-sub
}

Puedes usar los siguientes ejemplos de código para implementar este patrón en el lenguaje que elijas:

Go

En el siguiente ejemplo, se usa la versión principal de la biblioteca cliente de Pub/Sub para Go (v2). Si aún usas la biblioteca v1, consulta la guía de migración a la v2. Para ver una lista de ejemplos de código de la v1, consulta los ejemplos de código obsoletos.

Antes de probar esta muestra, sigue las instrucciones de configuración de Go en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Go.

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

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// optimisticSubscribe shows the recommended pattern for optimistically
// assuming a subscription exists prior to receiving messages.
func optimisticSubscribe(w io.Writer, projectID, topic, subscriptionName string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
	// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
	// If a subscription ID is provided, the project ID from the client is used.
	sub := client.Subscriber(subscriptionName)

	// 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()

	// Instead of checking if the subscription exists, optimistically try to
	// receive from the subscription assuming it exists.
	err = sub.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
		fmt.Fprintf(w, "Got from existing subscription: %q\n", string(msg.Data))
		msg.Ack()
	})
	if err != nil {
		if st, ok := status.FromError(err); ok {
			if st.Code() == codes.NotFound {
				// If the subscription does not exist, then create the subscription.
				subscription, err := client.SubscriptionAdminClient.CreateSubscription(ctx, &pubsubpb.Subscription{
					Name:  subscriptionName,
					Topic: topic,
				})
				if err != nil {
					return err
				}
				fmt.Fprintf(w, "Created subscription: %q\n", subscriptionName)

				// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
				// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
				// If a subscription ID is provided, the project ID from the client is used.
				sub = client.Subscriber(subscription.GetName())
				err = sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
					fmt.Fprintf(w, "Got from new subscription: %q\n", string(msg.Data))
					msg.Ack()
				})
				if err != nil && !errors.Is(err, context.Canceled) {
					return err
				}
			}
		}
	}
	return nil
}

Java

Antes de probar esta muestra, sigue las instrucciones de configuración de Java en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Java .

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.PushConfig;
import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class OptimisticSubscribeExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String subscriptionId = "your-subscription-id";
    String topicId = "your-topic-id";

    optimisticSubscribeExample(projectId, subscriptionId, topicId);
  }

  public static void optimisticSubscribeExample(
      String projectId, String subscriptionId, String topicId) throws IOException {
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        (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());
          consumer.ack();
        };

    Subscriber subscriber = null;
    try {
      subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();

      // Listen for resource NOT_FOUND errors and rebuild the  subscriber and restart subscribing
      // when the current subscriber encounters these errors.
      subscriber.addListener(
          new Subscriber.Listener() {
            public void failed(Subscriber.State from, Throwable failure) {
              System.out.println(failure.getStackTrace());
              if (failure instanceof NotFoundException) {
                try (SubscriptionAdminClient subscriptionAdminClient =
                    SubscriptionAdminClient.create()) {
                  TopicName topicName = TopicName.of(projectId, topicId);
                  // Create a pull subscription with default acknowledgement deadline of 10 seconds.
                  // The client library will automatically extend acknowledgement deadlines.
                  Subscription subscription =
                      subscriptionAdminClient.createSubscription(
                          subscriptionName, topicName, PushConfig.getDefaultInstance(), 10);
                  System.out.println("Created pull subscription: " + subscription.getName());
                  optimisticSubscribeExample(projectId, subscriptionId, topicId);
                } catch (IOException err) {
                  System.out.println("Failed to create pull subscription: " + err.getMessage());
                }
              }
            }
          },
          MoreExecutors.directExecutor());

      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (IllegalStateException e) {
      // Prevent an exception from being thrown if it is the expected NotFoundException
      if (!(subscriber.failureCause() instanceof NotFoundException)) {
        throw e;
      }
    } catch (TimeoutException e) {
      subscriber.stopAsync();
    }
  }
}

Node.js

Antes de probar esta muestra, sigue las instrucciones de configuración de Node.js en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Node.js.

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

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

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

function optimisticSubscribe(subscriptionNameOrId, topicNameOrId, timeout) {
  // Try using an existing subscription
  let subscription = pubSubClient.subscription(subscriptionNameOrId);

  // Create an event handler to handle messages
  let messageCount = 0;
  const messageHandler = message => {
    console.log(`Received message ${message.id}:`);
    console.log(`\tData: ${message.data}`);
    console.log(`\tAttributes: ${message.attributes}`);
    messageCount += 1;

    // "Ack" (acknowledge receipt of) the message
    message.ack();
  };

  // Set an error handler so that we're notified if the subscription doesn't
  // already exist.
  subscription.on('error', async e => {
    // Resource Not Found
    if (e.code === 5) {
      console.log('Subscription not found, creating it');
      await pubSubClient.createSubscription(
        topicNameOrId,
        subscriptionNameOrId,
      );

      // Refresh our subscriber object and re-attach the message handler.
      subscription = pubSubClient.subscription(subscriptionNameOrId);
      subscription.on('message', messageHandler);
    }
  });

  // Listen for new messages until timeout is hit; this will attempt to
  // open the actual subscriber streams. If it fails, the error handler
  // above will be called.
  subscription.on('message', messageHandler);

  // Wait a while for the subscription to run. (Part of the sample only.)
  setTimeout(() => {
    subscription.removeListener('message', messageHandler);
    console.log(`${messageCount} message(s) received.`);
  }, timeout * 1000);
}

Node.ts

Antes de probar esta muestra, sigue las instrucciones de configuración de Node.js en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Node.js.

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

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

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

function optimisticSubscribe(
  subscriptionNameOrId: string,
  topicNameOrId: string,
  timeout: number,
) {
  // Try using an existing subscription
  let subscription = pubSubClient.subscription(subscriptionNameOrId);

  // Create an event handler to handle messages
  let messageCount = 0;
  const messageHandler = (message: Message) => {
    console.log(`Received message ${message.id}:`);
    console.log(`\tData: ${message.data}`);
    console.log(`\tAttributes: ${message.attributes}`);
    messageCount += 1;

    // "Ack" (acknowledge receipt of) the message
    message.ack();
  };

  // Set an error handler so that we're notified if the subscription doesn't
  // already exist.
  subscription.on('error', async (e: StatusError) => {
    // Resource Not Found
    if (e.code === 5) {
      console.log('Subscription not found, creating it');
      await pubSubClient.createSubscription(
        topicNameOrId,
        subscriptionNameOrId,
      );

      // Refresh our subscriber object and re-attach the message handler.
      subscription = pubSubClient.subscription(subscriptionNameOrId);
      subscription.on('message', messageHandler);
    }
  });

  // Listen for new messages until timeout is hit; this will attempt to
  // open the actual subscriber streams. If it fails, the error handler
  // above will be called.
  subscription.on('message', messageHandler);

  // Wait a while for the subscription to run. (Part of the sample only.)
  setTimeout(() => {
    subscription.removeListener('message', messageHandler);
    console.log(`${messageCount} message(s) received.`);
  }, timeout * 1000);
}

Python

Antes de probar esta muestra, sigue las instrucciones de configuración de Python en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Python de Pub/Sub .

from google.api_core.exceptions import NotFound
from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError

# TODO(developer)
# project_id = "your-project-id"
# subscription_id = "your-subscription-id"
# Number of seconds the subscriber should listen for messages
# timeout = 5.0
# topic_id = "your-topic-id"

# Create a subscriber client.
subscriber = pubsub_v1.SubscriberClient()

# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)

# Define callback to be called when a message is received.
def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    # Ack message after processing it.
    message.ack()

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # Optimistically subscribe to messages on the subscription.
        streaming_pull_future = subscriber.subscribe(
            subscription_path, callback=callback
        )
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        print("Successfully subscribed until the timeout passed.")
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.
    except NotFound:
        print(f"Subscription {subscription_path} not found, creating it.")

        try:
            # If the subscription does not exist, then create it.
            publisher = pubsub_v1.PublisherClient()
            topic_path = publisher.topic_path(project_id, topic_id)
            subscription = subscriber.create_subscription(
                request={"name": subscription_path, "topic": topic_path}
            )

            if subscription:
                print(f"Subscription {subscription.name} created")
            else:
                raise ValueError("Subscription creation failed.")

            # Subscribe on the created subscription.
            try:
                streaming_pull_future = subscriber.subscribe(
                    subscription.name, callback=callback
                )
                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.
        except Exception as e:
            print(
                f"Exception occurred when creating subscription and subscribing to it: {e}"
            )
    except Exception as e:
        print(f"Exception occurred when attempting optimistic subscribe: {e}")

C++

Antes de probar esta muestra, sigue las instrucciones de configuración de C++ en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para C++ .

auto process_response = [](gc::StatusOr<pubsub::PullResponse> response) {
  if (response) {
    std::cout << "Received message " << response->message << "\n";
    std::move(response->handler).ack();
    return gc::Status();
  }
  if (response.status().code() == gc::StatusCode::kUnavailable &&
      response.status().message() == "no messages returned") {
    std::cout << "No messages returned from Pull()\n";
    return gc::Status();
  }
  return response.status();
};

// Instead of checking if the subscription exists, optimistically try to
// consume from the subscription.
auto status = process_response(subscriber.Pull());
if (status.ok()) return;
if (status.code() != gc::StatusCode::kNotFound) throw std::move(status);

// Since the subscription does not exist, create the subscription.
pubsub_admin::SubscriptionAdminClient subscription_admin_client(
    pubsub_admin::MakeSubscriptionAdminConnection());
google::pubsub::v1::Subscription request;
request.set_name(
    pubsub::Subscription(project_id, subscription_id).FullName());
request.set_topic(
    pubsub::Topic(project_id, std::move(topic_id)).FullName());
auto sub = subscription_admin_client.CreateSubscription(request);
if (!sub) throw std::move(sub).status();

// Consume from the new subscription.
status = process_response(subscriber.Pull());
if (!status.ok()) throw std::move(status);