GCP PubSub

This page describes the usage of the Stream Reactor Google PubSub Source Connector.

The Kafka connector is designed to seamlessly ingest records from GCP Pub/Sub topics and queues into your Kafka cluster. This makes it useful for backing up or streaming data from Pub/Sub to your Kafka infrastructure. This connector provides robust support for at least once semantics (this connector ensures that each record reaches the Kafka topic at least once).

Connector Class

io.lenses.streamreactor.connect.gcp.pubsub.source.GCPPubSubSourceConnector

Example

For more examples see the tutorials.

name=GcpPubSubSourceDemo
connector.class=io.lenses.streamreactor.connect.gcp.pubsub.source.GCPPubSubSourceConnector
topics=kafka_topic_to_write_to
tasks.max=1
connect.pubsub.gcp.auth.mode=File
connect.pubsub.gcp.file=/path/to/gcp-service-account-key.json
connect.pubsub.gcp.project.id=gcp-project-id
connect.pubsub.kcql=insert into `kafka_topic_to_write_to` select * from `gcp-subscription-id`

KCQL Support

The connector uses a SQL-like syntax to configure the connector behaviour. The full KCQL syntax is:

INSERT INTO kafka-topic
SELECT *
FROM subscriptionId
[PROPERTIES(
  'property.1'=x,
  'property.2'=x,
)]

Please note that you can employ escaping within KCQL for the INSERT INTO and SELECT * FROM clauses when necessary. For example, if you need to use a topic name that contains a hyphen, you can escape it as follows:

INSERT INTO `my-topic-with-hyphen`
SELECT *
FROM bucketAddress:pathPrefix

Source Subscription ID and Target Topic

The source and target of the data are specified via the INSERT INTO... SELECT * FROM clause. The connector will write all the records to the given topic, from the given subscription:

INSERT INTO my-topic SELECT * FROM subscriptionId;

Properties

The PROPERTIES clause is optional and adds a layer of configurability to the connector. It enhances versatility by permitting the application of multiple configurations (delimited by ',').

Properties can be defined in any order.

The following properties are supported:

Name	Description	Type	Default Value
batch.size	The maximum number of messages the connector will retrieve and process at one time per polling request (per KCQL mapping).	int	1000
cache.ttl	The maximum amount of time (in milliseconds) to store message data to allow acknowledgement of a message.	long	1 hour
queue.max	Data is loaded into a queue asynchronously so that it stands ready when the poll call is activated. Control the maximum number of records to hold in the queue per KCQL mapping.	int	10000

Auth Mode

The connector offers three distinct authentication modes:

• Default: This mode relies on the default GCP authentication chain, simplifying the authentication process.

• File: This mode uses a local (to the connect worker) path for a file containing GCP authentication credentials.

• Credentials: In this mode, explicit configuration of a GCP Credentials string is required for authentication.

The simplest example to configure in the connector is the "Default" mode, as this requires no other configuration.

connect.pubsub.gcp.auth.mode=Default

When selecting the "Credentials" mode, it is essential to provide the necessary credentials. Alternatively, if you prefer not to configure these properties explicitly, the connector will follow the credentials retrieval order as described here.

Here's an example configuration for the "Credentials" mode:

connect.pubsub.gcp.auth.mode=Credentials
connect.pubsub.gcp.credentials=$GCP_CREDENTIALS
connect.pubsub.gcp.project.id=$GCP_PROJECT_ID

And here is an example configuration using the "File" mode:

connect.pubsub.gcp.auth.mode=File
connect.pubsub.gcp.file=/home/secure-stuff/gcp-read-credential.txt

Remember when using file mode the file will need to exist on every worker node in your Kafka connect cluster and be readable by the Kafka Connect process.

For enhanced security and flexibility when using either the "Credentials" mode, it is highly advisable to utilize Connect Secret Providers.

Output Modes

Two modes are available: Default Mode and Compatibility Mode.

Compatibility Mode is intended to ensure compatibility with existing tools, while Default Mode offers a simpler modern redesign of the functionality.

You can choose whichever suits your requirements.

Default Mode

Configuration

connect.pubsub.output.mode=DEFAULT

Record Schema

Each Pub/Sub message is transformed into a single Kafka record, structured as follows:

• Kafka Key: A String of the Pub/Sub MessageID.

• Kafka Value: The Pub/Sub message value as BYTES.

• Kafka Headers: Includes the "PublishTimestamp" (in seconds) and all Pub/Sub message attributes mapped as separate headers.

Key Schema

The Kafka Key is mapped from the Pub/Sub MessageID, a unique ID for a Pub/Sub message.

Value Schema

The Kafka Value is mapped from the body of the Pub/Sub message.

Headers Schema

The Kafka Headers include:

• PublishTimestamp: Long value representing the time when the Pub/Sub message was published, in seconds.

• GCPProjectID: The GCP Project

• PubSubTopicID: The Pub/Sub Topic ID.

• PubSubSubscriptionID: The Pub/Sub Subscription ID.

• All Pub/Sub message attributes: Each attribute from the Pub/Sub message is mapped as a separate header.

Compatibility Mode

Configuration

connect.pubsub.output.mode=COMPATIBILITY

Record Schema

Each Pub/Sub message is transformed into a single Kafka record, structured as follows:

• Kafka Key: Comprises the project ID, message ID, and subscription ID of the Pub/Sub message.

• Kafka Value: Contains the message data and attributes from the Pub/Sub message.

Key Schema

The Key is a structure with these fields:

Field Name	Schema Type	Description
ProjectId	String	The Pub/Sub project containing the topic from which messages are polled.
TopicId	String	The Pub/Sub topic containing the messages.
SubscriptionId	String	The Pub/Sub subscription of the Pub/Sub topic.
MessageId	String	A unique ID for a Pub/Sub message

Value Schema

The Value is a structure with these fields:

Field Name	Schema Type	Description
MessageData	Optional String	The body of the Pub/Sub message.
AttributeMap	Optional String	The attribute map associated with the Pub/Sub message.

Option Reference

Name	Description	Type	Available Values	Default Value
connect.pubsub.gcp.auth.mode	Specifies the authentication mode for connecting to GCP.	string	Credentials, File or Default	Default
connect.pubsub.gcp.credentials	For “auth.mode” credentials: GCP Authentication credentials string.	string		(Empty)
connect.pubsub.gcp.file	For “auth.mode” file: Local file path for file containing GCP authentication credentials.	string		(Empty)
connect.pubsub.gcp.project.id	GCP Project ID.	string		(Empty)
connect.pubsub.kcql	Kafka Connect Query Language (KCQL) Configuration to control the connector behaviour	string	kcql configuration
connect.pubsub.output.mode	Output mode. Please see Output Modes documentation below.		Default or Compatibility	Default