Google Cloud Pub/Sub #

Among all the managed message brokers available in the cloud, Google Cloud Pub/Sub holds a unique position: it was designed from the start for global scale, not regional scale. A single Pub/Sub topic can receive and distribute messages to subscribers anywhere in the world without you needing to think about replication or geo-routing. This makes it the natural choice for applications built on Google Cloud Platform — from data analytics pipelines tightly integrated with BigQuery, to event-driven architectures connecting Cloud Run, Cloud Functions, and Dataflow. In terms of its model, Pub/Sub is more similar to a combination of SNS and SQS on AWS: one topic can have many subscriptions, and each subscription acts like its own queue receiving a copy of all the topic’s messages.

Pub/Sub Architecture #

Understanding Pub/Sub’s model is very important before writing code, because some of its concepts feel counterintuitive if you’re used to other brokers.

Topics and Subscriptions #

Topic is the resource where publishers send messages. A topic itself doesn’t store messages — it’s just a distribution channel.

Subscription is a resource attached to a topic that actually stores messages until they’re acknowledged or until the retention period expires. One topic can have many subscriptions, and each subscription independently receives a copy of all messages — exactly like a consumer group in Kafka, but with a different working mechanism.

flowchart LR
    PUB[Publisher] --> T["Topic:\norder-events"]

    T --> S1["Subscription:\nanalytics-sub"]
    T --> S2["Subscription:\nnotification-sub"]
    T --> S3["Subscription:\naudit-sub"]

    S1 --> C1["Analytics Service\nPull subscriber"]
    S2 --> C2["Notification Service\nPull subscriber"]
    S3 --> WH["Webhook endpoint\nPush subscriber"]

A critical point that often confuses people: if no subscription exists when a message is published, that message is lost. Pub/Sub doesn’t store messages at the topic level, only at the subscription level. Create the subscription before publishers start sending messages.

Push vs Pull Delivery #

Pub/Sub supports two delivery mechanisms to subscribers:

Pull — the subscriber actively requests messages from Pub/Sub. Similar to long polling in SQS. Suitable for consumers running as long-running processes (Kubernetes pods, VMs, always-on Cloud Run).

Push — Pub/Sub actively sends messages to the subscriber’s HTTPS endpoint. Suitable for Cloud Functions, Cloud Run (serverless), or webhooks. Pub/Sub automatically retries if the endpoint returns a non-2xx response.

flowchart TD
    T[Topic]

    subgraph PULL[Pull Subscription]
        S1[Subscription] -->|subscriber pulls messages| C1["Consumer\nlong-running service"]
    end

    subgraph PUSH[Push Subscription]
        S2[Subscription] -->|Pub/Sub pushes messages| EP["HTTPS Endpoint\nCloud Function / webhook"]
    end

    T --> S1
    T --> S2

	Pull	Push
Who initiates	Subscriber	Pub/Sub
Suitable for	Long-running services	Serverless, webhooks
Rate control	Subscriber controls itself	Pub/Sub controls (max burst rate)
Authentication	IAM on the subscriber side	OIDC token in the request header
Operations	Need to manage a polling loop	Zero ops on the consumer side

Acknowledgment Deadline #

Similar to the visibility timeout in SQS. When a subscriber receives a message via pull, it has the acknowledgment deadline period (default 10 seconds, max 600 seconds) to send an acknowledge. If not, Pub/Sub redelivers the message.

Message received by the subscriber
    │
    ├── within the ack deadline → subscriber sends ack → message removed from the subscription
    │
    └── past the ack deadline → Pub/Sub redelivers to another subscriber (or the same one)

Setting Up Dependencies #

Add the Google Cloud Pub/Sub client library to your Java project.

For Maven:

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.google.cloud</groupId>
            <artifactId>libraries-bom</artifactId>
            <version>26.37.0</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <!-- Google Cloud Pub/Sub -->
    <dependency>
        <groupId>com.google.cloud</groupId>
        <artifactId>google-cloud-pubsub</artifactId>
    </dependency>
</dependencies>

For Gradle:

dependencies {
    implementation platform('com.google.cloud:libraries-bom:26.37.0')
    implementation 'com.google.cloud:google-cloud-pubsub'
}

For local development, use the Pub/Sub Emulator:

# Install the Google Cloud SDK if you don't have it
# Then run the emulator
gcloud beta emulators pubsub start --project=my-project --host-port=localhost:8085

# Set environment variables so the client library points to the emulator
export PUBSUB_EMULATOR_HOST=localhost:8085
export GOOGLE_CLOUD_PROJECT=my-project

Or use Docker:

docker run -d \
  --name pubsub-emulator \
  -p 8085:8085 \
  gcr.io/google.com/cloudsdktool/google-cloud-cli \
  gcloud beta emulators pubsub start \
    --project=my-project \
    --host-port=0.0.0.0:8085

Authentication #

Pub/Sub uses Google Cloud IAM for authentication. The client library automatically looks up Application Default Credentials (ADC) from several sources in order:

1. GOOGLE_APPLICATION_CREDENTIALS environment variable → path to a service account JSON
2. gcloud auth application-default login → credentials from the gcloud CLI
3. Metadata server → when running on GCE, GKE, Cloud Run, or Cloud Functions (IAM role)

import com.google.auth.oauth2.GoogleCredentials;
import com.google.auth.oauth2.ServiceAccountCredentials;

import java.io.FileInputStream;

public class PubSubAuth {

    // ✓ CORRECT: Application Default Credentials — automatic on GCP, gcloud for dev
    // No extra code needed — the library handles it
    // Just set GOOGLE_APPLICATION_CREDENTIALS or run gcloud auth

    // ✓ CORRECT: explicit service account if you need specific credentials
    public static GoogleCredentials loadServiceAccount(String keyFilePath) throws Exception {
        try (FileInputStream stream = new FileInputStream(keyFilePath)) {
            return ServiceAccountCredentials.fromStream(stream)
                .createScoped("https://www.googleapis.com/auth/cloud-platform");
        }
    }

    // ✗ ANTI-PATTERN: hardcoding a service account key in source code
    // The key could leak into version control or application logs
    public static final String HARDCODED_KEY = "{ \"type\": \"service_account\", ... }";
}

Don’t commit service account JSON keys to version control. Use Secret Manager, environment variables injected at deploy time, or Workload Identity (for GKE), which eliminates the need for service account keys entirely.

Managing Topics and Subscriptions #

Pub/Sub provides TopicAdminClient and SubscriptionAdminClient for programmatic resource management.

import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.*;

import org.threeten.bp.Duration;

public class PubSubResourceManager {

    private final String projectId;

    public PubSubResourceManager(String projectId) {
        this.projectId = projectId;
    }

    // Create a topic
    public void createTopic(String topicId) throws Exception {
        try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
            TopicName topicName = TopicName.of(projectId, topicId);

            Topic topic = Topic.newBuilder()
                .setName(topicName.toString())
                // Message retention at the topic level (not subscription) — 1 day
                // This lets subscriptions created after messages were published
                // access messages within this window (seek to timestamp)
                .setMessageRetentionDuration(
                    com.google.protobuf.Duration.newBuilder()
                        .setSeconds(86400) // 24 hours
                        .build()
                )
                .build();

            topicAdminClient.createTopic(topic);
            System.out.println("Topic created: " + topicName);
        }
    }

    // Create a Pull Subscription
    public void createPullSubscription(String topicId, String subscriptionId) throws Exception {
        try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
            TopicName topicName = TopicName.of(projectId, topicId);
            SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);

            Subscription subscription = Subscription.newBuilder()
                .setName(subscriptionName.toString())
                .setTopic(topicName.toString())
                .setAckDeadlineSeconds(60)         // the subscriber has 60 seconds to ack
                .setRetainAckedMessages(false)     // delete acknowledged messages
                .setMessageRetentionDuration(
                    com.google.protobuf.Duration.newBuilder()
                        .setSeconds(7 * 86400)     // keep unacked messages for 7 days
                        .build()
                )
                .build();

            subscriptionAdminClient.createSubscription(subscription);
            System.out.println("Pull subscription created: " + subscriptionName);
        }
    }

    // Create a Push Subscription
    public void createPushSubscription(String topicId, String subscriptionId,
                                        String pushEndpoint) throws Exception {
        try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
            TopicName topicName = TopicName.of(projectId, topicId);
            SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);

            PushConfig pushConfig = PushConfig.newBuilder()
                .setPushEndpoint(pushEndpoint) // must be HTTPS and publicly accessible
                .build();

            Subscription subscription = Subscription.newBuilder()
                .setName(subscriptionName.toString())
                .setTopic(topicName.toString())
                .setPushConfig(pushConfig)
                .setAckDeadlineSeconds(30)
                .build();

            subscriptionAdminClient.createSubscription(subscription);
            System.out.println("Push subscription created: " + subscriptionName);
        }
    }

    // Create a Subscription with a Dead Letter Topic
    public void createSubscriptionWithDLT(String topicId, String subscriptionId,
                                           String deadLetterTopicId) throws Exception {
        try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
            TopicName topicName = TopicName.of(projectId, topicId);
            TopicName deadLetterTopicName = TopicName.of(projectId, deadLetterTopicId);
            SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);

            DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.newBuilder()
                .setDeadLetterTopic(deadLetterTopicName.toString())
                .setMaxDeliveryAttempts(5) // send to the DLT after 5 failed attempts
                .build();

            Subscription subscription = Subscription.newBuilder()
                .setName(subscriptionName.toString())
                .setTopic(topicName.toString())
                .setDeadLetterPolicy(deadLetterPolicy)
                .setAckDeadlineSeconds(60)
                .build();

            subscriptionAdminClient.createSubscription(subscription);
            System.out.println("Subscription with DLT created: " + subscriptionName);
        }
    }
}

Publishers #

Publishers use the Publisher client, which supports asynchronous sending with automatic batching.

Publisher Configuration #

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.batching.BatchingSettings;
import com.google.api.gax.retrying.RetrySettings;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;

import org.threeten.bp.Duration;

import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class PubSubPublisher {

    private final Publisher publisher;

    public PubSubPublisher(String projectId, String topicId) throws Exception {
        TopicName topicName = TopicName.of(projectId, topicId);

        // Batching configuration — the library collects messages before sending
        BatchingSettings batchingSettings = BatchingSettings.newBuilder()
            .setElementCountThreshold(100L)         // send when 100 messages accumulate
            .setRequestByteThreshold(1024 * 1024L)  // or when the total reaches 1 MB
            .setDelayThreshold(Duration.ofMillis(100)) // or 100ms after the first message
            .build();

        // Retry configuration for transient errors
        RetrySettings retrySettings = RetrySettings.newBuilder()
            .setMaxAttempts(5)
            .setInitialRetryDelay(Duration.ofMillis(100))
            .setMaxRetryDelay(Duration.ofSeconds(60))
            .setRetryDelayMultiplier(2.0)
            .build();

        this.publisher = Publisher.newBuilder(topicName)
            .setBatchingSettings(batchingSettings)
            .setRetrySettings(retrySettings)
            .build();
    }

    // Publish a message with attributes (metadata)
    public void publish(String data, Map<String, String> attributes) {
        ByteString byteData = ByteString.copyFromUtf8(data);

        PubsubMessage.Builder messageBuilder = PubsubMessage.newBuilder()
            .setData(byteData);

        // Add attributes — key-value metadata that can be used for filtering
        if (attributes != null) {
            messageBuilder.putAllAttributes(attributes);
        }

        PubsubMessage message = messageBuilder.build();

        // publish() returns an ApiFuture — non-blocking
        ApiFuture<String> future = publisher.publish(message);

        // Register a callback to learn the outcome
        ApiFutures.addCallback(future, new ApiFutureCallback<String>() {
            @Override
            public void onSuccess(String messageId) {
                System.out.println("Message published successfully — MessageId: " + messageId);
            }

            @Override
            public void onFailure(Throwable throwable) {
                System.err.println("Failed to publish message: " + throwable.getMessage());
                // Here: log the error, send to a fallback, or alert monitoring
            }
        }, Executors.newSingleThreadExecutor());
    }

    // Publish with an ordering key — requires message ordering enabled on the subscription
    public void publishOrdered(String data, String orderingKey) {
        PubsubMessage message = PubsubMessage.newBuilder()
            .setData(ByteString.copyFromUtf8(data))
            .setOrderingKey(orderingKey) // all messages with the same key are delivered in order
            .build();

        publisher.publish(message);
    }

    // Must be called when the application shuts down
    // The Publisher flushes all buffered messages before stopping
    public void shutdown() throws Exception {
        publisher.shutdown();
        publisher.awaitTermination(30, TimeUnit.SECONDS);
    }
}

Explicit Batch Publishing #

Even though batching is done automatically by the library, you can publish many messages at once and wait for all of them to finish:

import java.util.ArrayList;
import java.util.List;

public class BatchPublisher {

    private final Publisher publisher;

    public BatchPublisher(Publisher publisher) {
        this.publisher = publisher;
    }

    public void publishBatch(List<String> messages) throws Exception {
        List<ApiFuture<String>> futures = new ArrayList<>();

        for (String message : messages) {
            PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
                .setData(ByteString.copyFromUtf8(message))
                .build();

            futures.add(publisher.publish(pubsubMessage));
        }

        // Wait for all messages to finish publishing
        // ApiFutures.allAsList() fails if any one of them fails
        try {
            List<String> messageIds = ApiFutures.allAsList(futures).get();
            System.out.printf("Successfully published %d messages%n", messageIds.size());
        } catch (Exception e) {
            System.err.println("One or more messages failed to publish: " + e.getMessage());
            throw e;
        }
    }
}

Subscribers — Pull #

Pull subscribers use the Subscriber client, which handles polling, threading, and acking automatically in the background.

Synchronous Pull #

Use synchronous pull for scenarios where you need full control over when messages are processed — for example, batch pipelines that only run at certain times:

import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub;
import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings;
import com.google.pubsub.v1.*;

public class SynchronousPullSubscriber {

    private final String projectId;
    private final String subscriptionId;

    public SynchronousPullSubscriber(String projectId, String subscriptionId) {
        this.projectId = projectId;
        this.subscriptionId = subscriptionId;
    }

    public void pullAndProcess(int maxMessages) throws Exception {
        SubscriberStubSettings subscriberStubSettings = SubscriberStubSettings.newBuilder()
            .setTransportChannelProvider(
                SubscriberStubSettings.defaultGrpcTransportProviderBuilder()
                    .setMaxInboundMessageSize(20 * 1024 * 1024) // 20 MB
                    .build()
            )
            .build();

        try (GrpcSubscriberStub subscriber = GrpcSubscriberStub.create(subscriberStubSettings)) {
            String subscriptionName = SubscriptionName.format(projectId, subscriptionId);

            PullRequest pullRequest = PullRequest.newBuilder()
                .setMaxMessages(maxMessages)
                .setSubscription(subscriptionName)
                .build();

            PullResponse pullResponse = subscriber.pullCallable().call(pullRequest);
            List<String> ackIds = new ArrayList<>();

            for (ReceivedMessage receivedMessage : pullResponse.getReceivedMessagesList()) {
                PubsubMessage message = receivedMessage.getMessage();
                String data = message.getData().toStringUtf8();

                System.out.printf("MessageId=%s | Data: %s%n",
                    message.getMessageId(), data);
                System.out.println("Attributes: " + message.getAttributesMap());

                try {
                    processMessage(data, message.getAttributesMap());
                    ackIds.add(receivedMessage.getAckId());
                } catch (Exception e) {
                    System.err.println("Failed to process, won't ack: " + e.getMessage());
                    // Don't add to ackIds — Pub/Sub will redeliver after the deadline
                }
            }

            // Acknowledge all successfully processed messages in one request
            if (!ackIds.isEmpty()) {
                AcknowledgeRequest acknowledgeRequest = AcknowledgeRequest.newBuilder()
                    .setSubscription(subscriptionName)
                    .addAllAckIds(ackIds)
                    .build();

                subscriber.acknowledgeCallable().call(acknowledgeRequest);
                System.out.println("Acknowledged " + ackIds.size() + " messages.");
            }
        }
    }

    private void processMessage(String data, Map<String, String> attributes) throws Exception {
        // business logic
    }
}

Asynchronous Pull (Streaming) #

For long-running consumers, use streaming pull with Subscriber — this is the most common and efficient way:

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 StreamingPullSubscriber {

    public static void startSubscriber(String projectId, String subscriptionId) throws Exception {
        ProjectSubscriptionName subscriptionName =
            ProjectSubscriptionName.of(projectId, subscriptionId);

        // MessageReceiver is called on a separate thread for each message
        MessageReceiver receiver = (PubsubMessage message, AckReplyConsumer consumer) -> {
            String data = message.getData().toStringUtf8();
            String messageId = message.getMessageId();
            Map<String, String> attributes = message.getAttributesMap();

            System.out.printf("Received message — MessageId=%s%n", messageId);

            try {
                processMessage(data, attributes);

                // ✓ Ack — Pub/Sub removes the message from the subscription
                consumer.ack();
                System.out.println("Message acked: " + messageId);

            } catch (Exception e) {
                System.err.printf("Failed to process MessageId=%s: %s%n", messageId, e.getMessage());

                // ✓ Nack — Pub/Sub redelivers after the ack deadline
                // If the subscription has a Dead Letter Policy, the message enters the DLT
                // after maxDeliveryAttempts is reached
                consumer.nack();
            }
        };

        Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver)
            .setMaxAckExtensionPeriod(Duration.ofSeconds(120)) // extend the ack deadline automatically
            .setParallelPullCount(2)      // number of parallel streaming pulls
            .setExecutorProvider(         // thread pool for processing messages
                com.google.api.gax.core.InstantiatingExecutorProvider.newBuilder()
                    .setExecutorThreadCount(4)
                    .build()
            )
            .build();

        // Register a listener for fatal errors
        subscriber.addListener(new Subscriber.Listener() {
            @Override
            public void failed(Subscriber.State from, Throwable failure) {
                System.err.println("Subscriber failed from state " + from + ": " + failure.getMessage());
                // Alert the monitoring system
            }
        }, Executors.newSingleThreadExecutor());

        subscriber.startAsync().awaitRunning();
        System.out.println("Subscriber running, waiting for messages...");

        // Run forever (or until a shutdown signal)
        try {
            subscriber.awaitTerminated(30, TimeUnit.MINUTES);
        } catch (TimeoutException e) {
            subscriber.stopAsync();
        }
    }

    private static void processMessage(String data, Map<String, String> attributes) throws Exception {
        System.out.println("Processing: " + data);
        // business logic
    }
}

Extending the Acknowledgment Deadline #

If processing takes longer than the ack deadline, use ModifyAckDeadline to extend it:

// With the async Subscriber, the library handles this automatically via setMaxAckExtensionPeriod
// For synchronous pull, you need to do it manually:

public void extendDeadline(GrpcSubscriberStub subscriber,
                            String subscriptionName,
                            List<String> ackIds,
                            int newDeadlineSeconds) {
    ModifyAckDeadlineRequest modifyRequest = ModifyAckDeadlineRequest.newBuilder()
        .setSubscription(subscriptionName)
        .addAllAckIds(ackIds)
        .setAckDeadlineSeconds(newDeadlineSeconds) // extend by N seconds from now
        .build();

    subscriber.modifyAckDeadlineCallable().call(modifyRequest);
    System.out.println("Ack deadline extended by " + newDeadlineSeconds + " seconds.");
}

Message Filtering #

Pub/Sub supports CEL (Common Expression Language) based filters at the subscription level. Subscribers only receive messages matching the filter — non-matching messages are automatically acknowledged by Pub/Sub (they don’t enter the subscription).

// Filter for messages whose "event-type" attribute is "order.created"
// or "order.shipped"
String filter = "attributes.\"event-type\" = \"order.created\" " +
                "OR attributes.\"event-type\" = \"order.shipped\"";

Subscription subscription = Subscription.newBuilder()
    .setName(subscriptionName.toString())
    .setTopic(topicName.toString())
    .setFilter(filter) // the filter applies on the Pub/Sub side, not the subscriber
    .setAckDeadlineSeconds(60)
    .build();

subscriptionAdminClient.createSubscription(subscription);

Other common filter examples:

// Filter by a single attribute
attributes.region = "asia-southeast1"

// Filter by attribute existence
hasPrefix(attributes.order-id, "ORD-")

// Combined filters
attributes.env = "production" AND attributes.priority = "high"

// Filter by message data (must be JSON)
// Not supported — filters only work on message attributes, not the body

Pub/Sub filters only work on message attributes, not on the message body content. If you need content-based routing, do it on the consumer side after receiving the message, or encode routing criteria as attributes when publishing.

Message Ordering #

By default, Pub/Sub doesn’t guarantee message delivery order. To guarantee ordering, enable enableMessageOrdering on the Subscriber and use an orderingKey when publishing.

// Publisher — enable message ordering on the publisher
Publisher publisher = Publisher.newBuilder(topicName)
    .setEnableMessageOrdering(true) // required if you want to use ordering keys
    .build();

// Publish with an ordering key — all messages with the same key are delivered in order
PubsubMessage message = PubsubMessage.newBuilder()
    .setData(ByteString.copyFromUtf8(payload))
    .setOrderingKey("order-" + orderId) // messages for the same order are always ordered
    .build();

publisher.publish(message);

// Subscription — enable message ordering on the subscription
Subscription subscription = Subscription.newBuilder()
    .setName(subscriptionName.toString())
    .setTopic(topicName.toString())
    .setEnableMessageOrdering(true) // required so ordering keys are respected
    .setAckDeadlineSeconds(60)
    .build();

If a publisher fails to publish a message with a particular ordering key, all subsequent messages with the same ordering key will be rejected until you explicitly call publisher.resumePublish(orderingKey). This prevents gaps that would break the order.

Dead Letter Topics #

The Dead Letter Topic (DLT) in Pub/Sub works at the subscription level — messages that exceed the maxDeliveryAttempts limit without being acknowledged are automatically moved to another topic.

flowchart TD
    PUB[Publisher] --> T[Topic: order-events]
    T --> SUB["Subscription: order-sub\nmaxDeliveryAttempts=5"]
    SUB --> C{"Consumer\nacked successfully?"}

    C -- Yes --> DONE["Message removed\nfrom the subscription"]
    C -- No\nconsumer.nack / timeout --> SUB
    SUB -->|Failed 5 times| DLT["Dead Letter Topic:\norder-events-dlt"]
    DLT --> DLSUB[Subscription: dlq-sub]
    DLSUB --> MON["Monitoring &\nInvestigation"]

Messages entering the DLT include additional attributes for investigation:

CloudPubSubDeadLetterSourceSubscription → the name of the source subscription
CloudPubSubDeadLetterSourceTopicPublishTime → when the message was first published

A consumer for the DLT:

public class DeadLetterConsumer {

    public static void monitorDLT(String projectId, String dltSubscriptionId) throws Exception {
        ProjectSubscriptionName subscriptionName =
            ProjectSubscriptionName.of(projectId, dltSubscriptionId);

        MessageReceiver receiver = (PubsubMessage message, AckReplyConsumer consumer) -> {
            String data = message.getData().toStringUtf8();
            Map<String, String> attributes = message.getAttributesMap();

            // Get the message origin info from the attributes added by Pub/Sub
            String sourceSubscription = attributes.getOrDefault(
                "CloudPubSubDeadLetterSourceSubscription", "unknown"
            );

            System.err.printf(
                "[DLT] Failed message from subscription '%s': %s%n",
                sourceSubscription, data
            );

            // Send an alert and log for investigation
            sendAlert(data, attributes);

            // Ack in the DLT — the message is logged, no need to reprocess automatically
            consumer.ack();
        };

        Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
        subscriber.startAsync().awaitRunning();
        System.out.println("DLT consumer running...");
        subscriber.awaitTerminated();
    }

    private static void sendAlert(String data, Map<String, String> attributes) {
        System.out.println("Alert sent to the ops team: " + data);
    }
}

Seek — Replaying or Skipping Messages #

One feature that distinguishes Pub/Sub from SQS is seek — you can “rewind” a subscription to a specific point in the past, or skip all messages currently in the queue. This is useful for reprocessing after a bug is fixed or skipping an irrelevant backlog.

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.SeekRequest;
import com.google.pubsub.v1.SubscriptionName;

import com.google.protobuf.Timestamp;
import java.time.Instant;

public class PubSubSeek {

    // Seek to a specific point in time — the subscription will redeliver all messages
    // published after this timestamp (within the message retention window)
    public static void seekToTime(String projectId, String subscriptionId,
                                   Instant targetTime) throws Exception {
        try (SubscriptionAdminClient adminClient = SubscriptionAdminClient.create()) {
            SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);

            Timestamp timestamp = Timestamp.newBuilder()
                .setSeconds(targetTime.getEpochSecond())
                .build();

            SeekRequest seekRequest = SeekRequest.newBuilder()
                .setSubscription(subscriptionName.toString())
                .setTime(timestamp)
                .build();

            adminClient.seek(seekRequest);
            System.out.println("Subscription seeked to: " + targetTime);
        }
    }

    // Skip all existing messages — the subscription only receives new messages after this
    public static void skipBacklog(String projectId, String subscriptionId) throws Exception {
        // Seek to "now" — all existing messages are considered processed
        seekToTime(projectId, subscriptionId, Instant.now());
        System.out.println("Backlog skipped. Only receiving new messages.");
    }
}

Push Subscriptions — Receiving Messages via HTTP #

For push subscriptions, Pub/Sub sends an HTTP POST to your endpoint. The request body contains the message in a JSON envelope format:

import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Base64;

public class PushSubscriptionHandler implements HttpHandler {

    private static final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void handle(HttpExchange exchange) throws IOException {
        if (!"POST".equals(exchange.getRequestMethod())) {
            exchange.sendResponseHeaders(405, -1);
            return;
        }

        try {
            // Read the request body
            byte[] requestBody = exchange.getRequestBody().readAllBytes();
            String body = new String(requestBody);

            // Parse the JSON envelope from Pub/Sub
            // Format: { "message": { "data": "<base64>", "attributes": {...}, "messageId": "..." },
            //           "subscription": "..." }
            var envelope = objectMapper.readTree(body);
            var messageNode = envelope.get("message");

            // The data is base64-encoded
            String encodedData = messageNode.get("data").asText();
            String data = new String(Base64.getDecoder().decode(encodedData));

            String messageId = messageNode.get("messageId").asText();

            System.out.printf("Push message received — MessageId=%s: %s%n", messageId, data);

            // Process the message
            processMessage(data);

            // Return 2xx to acknowledge the message to Pub/Sub
            // Pub/Sub considers delivery successful if the response is 2xx within the ack deadline
            exchange.sendResponseHeaders(200, -1);

        } catch (Exception e) {
            System.err.println("Failed to process push message: " + e.getMessage());

            // Return non-2xx to tell Pub/Sub to retry
            exchange.sendResponseHeaders(500, -1);
        } finally {
            exchange.close();
        }
    }

    private void processMessage(String data) throws Exception {
        System.out.println("Processing: " + data);
        // business logic
    }

    // Example of a simple HTTP server to receive pushes
    public static void main(String[] args) throws Exception {
        HttpServer server = HttpServer.create(new InetSocketAddress(8080), 0);
        server.createContext("/pubsub/push", new PushSubscriptionHandler());
        server.start();
        System.out.println("Push endpoint running on port 8080");
    }
}

When to Use Pub/Sub and When Not To #

USE GOOGLE PUB/SUB WHEN:
  ✓ You're already in the Google Cloud Platform ecosystem
  ✓ You need native fan-out — one topic, many independent subscribers
  ✓ Tight integration with BigQuery, Dataflow, Cloud Functions, Cloud Run
  ✓ You need global scale without geo-routing configuration
  ✓ You want zero ops — no broker to maintain
  ✓ Push subscribers for serverless or simple webhooks
  ✓ You need broker-side message filtering (based on attributes)
  ✓ You need seek — replaying or skipping historical messages

CONSIDER ALTERNATIVES WHEN:
  ✗ You're not on GCP and don't want vendor lock-in → Kafka or RabbitMQ
  ✗ You need complex routing (exchange patterns) → RabbitMQ
  ✗ You need very high throughput with ultra-low latency → Kafka
  ✗ You need global ordering (not per key) → Kafka with a single partition
  ✗ Messages larger than 10 MB → Pub/Sub doesn't support it, use GCS + pointer
  ✗ You need exactly-once end-to-end processing → Pub/Sub is only at-least-once

flowchart TD
    A{"Already in the\nGCP ecosystem?"} -- Yes --> B{"Need fan-out\nto many subscribers?"}
    A -- No --> C{"Need a managed\nservice?"}

    B -- Yes --> PUBSUB["Google Pub/Sub"]
    B -- No --> D{"Need GCP services\nintegration?"}
    D -- Yes --> PUBSUB
    D -- No --> E{"Need complex\nrouting?"}
    E -- Yes --> RABBIT[RabbitMQ]
    E -- No --> SQS[Amazon SQS]

    C -- Yes --> F{On AWS?}
    F -- Yes --> SQS
    F -- No --> PUBSUB

    C -- No --> G{"Very high volume /\nreplay?"}
    G -- Yes --> KAFKA[Kafka]
    G -- No --> RABBIT

Summary #

A topic is a distribution channel, a subscription stores messages — create the subscription before publishers start sending, or messages will be permanently lost.
Pull for long-running services, push for serverless — push subscriptions eliminate the need for polling loops, but the endpoint must be publicly accessible via HTTPS.
The acknowledgment deadline determines how long a subscriber has to process and ack a message. Use setMaxAckExtensionPeriod on the async Subscriber so the library extends the deadline automatically.
Message ordering requires enableMessageOrdering=true on both the publisher and subscription, plus an orderingKey on every message. If publishing fails, call resumePublish(orderingKey) before continuing.
Subscription filters enable attribute-based message routing on the Pub/Sub side — subscribers only receive relevant messages without needing application-side filtering.
The Dead Letter Topic is configured at the subscription level via maxDeliveryAttempts — cleaner than manual DLQ implementations because Pub/Sub handles message movement automatically.
Seek is a unique Pub/Sub feature — you can replay messages from a specific timestamp or skip an entire backlog without changing consumer code.
Pub/Sub fits best for the GCP ecosystem, fan-out to many subscribers, GCP services integration, and global scalability — not for complex routing or very large messages.

← Previous: Amazon SQS Next: Redis →