= Google Pubsub Component
:doctitle: Google Pubsub
:shortname: google-pubsub
:artifactid: camel-google-pubsub
:description: Send and receive messages to/from Google Cloud Platform PubSub Service.
:since: 2.19
:supportlevel: Stable
:tabs-sync-option:
:component-header: Both producer and consumer are supported
//Manually maintained attributes
:group: Google
:camel-spring-boot-name: google-pubsub

*Since Camel {since}*

*{component-header}*

The Google Pubsub component provides access
to https://cloud.google.com/pubsub/[Cloud Pub/Sub Infrastructure] via
the https://github.com/googleapis/java-pubsub[Google Cloud Java Client for Google Cloud Pub/Sub].

Maven users will need to add the following dependency to their pom.xml
for this component:

[source,xml]
------------------------------------------------------
<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-google-pubsub</artifactId>
    <!-- use the same version as your Camel core version -->
    <version>x.x.x</version>
</dependency>
------------------------------------------------------

== URI Format

The Google Pubsub Component uses the following URI format:

----
google-pubsub://project-id:destinationName?[options]
----

Destination Name can be either a topic or a subscription name.


// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
include::partial$component-endpoint-headers.adoc[]
// component options: END

== Usage

=== Producer Endpoints

Producer endpoints can accept and deliver to PubSub individual and grouped
exchanges alike. Grouped exchanges have `Exchange.GROUPED_EXCHANGE` property set.

Google PubSub expects the payload to be byte[] array, Producer endpoints will send:

* String body as `byte[]` encoded as UTF-8
* `byte[]` body as is
* Everything else will be serialised into a `byte[]` array

A Map set as message header `GooglePubsubConstants.ATTRIBUTES` will be sent as PubSub attributes.

Google PubSub supports ordered message delivery.

To enable this, set the options `messageOrderingEnabled` to true, and the `pubsubEndpoint` to a GCP region.

When producing messages set the message header `GooglePubsubConstants.ORDERING_KEY` . This will be set as the PubSub orderingKey for the message.

For more information, see https://cloud.google.com/pubsub/docs/ordering[Ordering messages].

Once exchange has been delivered to PubSub the PubSub Message ID will be assigned to
the header `GooglePubsubConstants.MESSAGE_ID`.

=== Consumer Endpoints
Google PubSub will redeliver the message if it has not been acknowledged within the time period set
as a configuration option on the subscription.

The component will acknowledge the message once exchange processing has been completed.

If the route throws an exception, the exchange is marked as failed, and the component will NACK the message.
It will be redelivered immediately.

To ack/nack the message the component uses Acknowledgement ID stored as header `GooglePubsubConstants.ACK_ID`.
If the header is removed or tampered with, the ack will fail and the message will be redelivered
again after the ack deadline.

=== Delivery Attempts and Dead Letter Queues

When a subscription has a https://cloud.google.com/pubsub/docs/dead-letter-topics[dead-letter policy] configured,
the component exposes the delivery attempt count via the header `GooglePubsubConstants.DELIVERY_ATTEMPT`.
This header contains an `Integer` representing the approximate number of times the message has been delivered.
The first delivery will have a value of `1`.

NOTE: The delivery attempt header is only available when the subscription has a dead-letter policy configured.
Without a dead-letter policy, the header will not be set.

This allows routes to implement custom retry logic based on the delivery attempt count:

[source,java]
----
from("google-pubsub:{{project.name}}:{{subscription.name}}")
    .process(exchange -> {
        Integer deliveryAttempt = exchange.getIn().getHeader(GooglePubsubConstants.DELIVERY_ATTEMPT, Integer.class);
        if (deliveryAttempt != null && deliveryAttempt > 3) {
            // Custom handling for messages that have been redelivered multiple times
            log.warn("Message has been delivered {} times", deliveryAttempt);
        }
        // Process the message
    });
----

With a dead-letter policy, after the configured maximum delivery attempts are exceeded,
the message will automatically be forwarded to the dead-letter topic by Google PubSub.

==== Automatic Max Delivery Attempts Enforcement

The component can enforce the subscription's `maxDeliveryAttempts` setting at the consumer level.
When enabled, messages whose delivery attempt count is greater than or equal to the configured maximum
will be automatically nacked without processing, allowing Pub/Sub to route them to the dead-letter topic.
This prevents infinite redelivery loops that can occur with short retry delays.

The `maxDeliveryAttempts` value is resolved as follows:

1. If explicitly set via the endpoint option, that value is used.
2. If not explicitly set, the component attempts to auto-fetch the value from the subscription's
   dead-letter policy at consumer startup.
3. If auto-fetch fails (e.g., insufficient permissions or no dead-letter policy), enforcement is
   disabled and a warning is logged.
4. A value of `0` disables enforcement.

[source,java]
----
// Explicit configuration
from("google-pubsub:{{project.name}}:{{subscription.name}}?maxDeliveryAttempts=5")
    .to("direct:process");

// Auto-fetch from subscription dead-letter policy (default behavior when not set)
from("google-pubsub:{{project.name}}:{{subscription.name}}")
    .to("direct:process");
----

=== Message Body

The consumer endpoint returns the content of the message as `byte[]`. Exactly as the underlying system sends it.
It is up for the route to convert/unmarshall the contents.

=== Authentication Configuration

By default, this component acquires credentials using `GoogleCredentials.getApplicationDefault()`.
This behavior can be disabled by setting _authenticate_ option to `false`, in which case requests to Google API will be made without authentication details. This is only desirable when developing against an emulator.
This behavior can be altered by supplying a path to a service account key file.

==== Workload Identity Federation (WIF)

All Google components support https://cloud.google.com/iam/docs/workload-identity-federation[Workload Identity Federation], which enables workloads running outside of Google Cloud (e.g., on AWS, Azure, GitHub Actions) or on GKE to authenticate without service account key files.

**On GKE with Workload Identity:** No configuration is needed. Application Default Credentials (ADC) automatically detects the GKE environment and uses the Kubernetes service account's associated GCP identity.

**With an explicit WIF configuration file:** Set `useWorkloadIdentityFederation=true` and provide the path to the WIF JSON config file via `workloadIdentityConfig`. This is the typical setup for GitHub Actions, AWS, and Azure workloads.

[source,java]
----
// GKE with Workload Identity - ADC handles it automatically
from("google-pubsub:my-project:my-subscription")
    .to("direct:process");

// GitHub Actions / AWS / Azure with WIF config file
GooglePubsubEndpoint endpoint = context.getEndpoint("google-pubsub:my-project:my-subscription", GooglePubsubEndpoint.class);
endpoint.setUseWorkloadIdentityFederation(true);
endpoint.setWorkloadIdentityConfig("/path/to/wif-config.json");
----

**With Service Account Impersonation:** Set `impersonatedServiceAccount` to a target service account email. The external credentials obtained via WIF will impersonate that service account, inheriting its permissions.

[source,java]
----
// WIF with service account impersonation
GooglePubsubEndpoint endpoint = context.getEndpoint("google-pubsub:my-project:my-subscription", GooglePubsubEndpoint.class);
endpoint.setUseWorkloadIdentityFederation(true);
endpoint.setWorkloadIdentityConfig("/path/to/wif-config.json");
endpoint.setImpersonatedServiceAccount("my-sa@my-project.iam.gserviceaccount.com");
----

NOTE: Workload Identity Federation support is available in all Google components (PubSub, Storage, BigQuery, Firestore, Sheets, Calendar, Drive, Mail, Functions, Secret Manager, Vision, Vertex AI, Speech-to-Text, Text-to-Speech) through the common `GoogleCommonConfiguration` interface.

include::spring-boot:partial$starter.adoc[]

=== Manual Acknowledgement

By default, the PubSub consumer will acknowledge messages once the exchange has been processed, or negative-acknowledge them if the exchange has failed.

If the _ackMode_ option is set to `NONE`, the component will not acknowledge messages, and it is up to the route to do so.
In this case, a `GooglePubsubAcknowledge` object is stored in the header `GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE` and can be used to acknowledge messages:

[source,java]
----
from("google-pubsub:{{project.name}}:{{subscription.name}}?ackMode=NONE")
    .process(exchange -> {
        GooglePubsubAcknowledge acknowledge = exchange.getIn().getHeader(GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE, GooglePubsubAcknowledge.class);
        acknowledge.ack(exchange); // or .nack(exchange)
    });
----

Manual acknowledgement works with both the asynchronous and synchronous consumers and will use the acknowledgement id which is stored in `GooglePubsubConstants.ACK_ID`.
