Google PubSub connector

• Startup
• Business
• Enterprise
• On-Premise
• Add-on

About Google PubSub

Pub/Sub is an asynchronous and scalable messaging service that decouples services producing messages from services processing those messages. Read more about Google Pub/Sub.

When to use this connector

• to read messages from Pub/Sub subscriptions and write messages to Pub/Sub topic(s).
• to stream messages from the queue to various destinations.
• to implement a log-based CDC with a message queue.
  • Stream CDC events to the message queue.
  • Stream CDC events from the message queue to any destination.

Create a Connection

Step 1. In the Connections window, click +, and type in google pubsub.

Step 2. Select Google PubSub.

Step 3. Enter the Connection parameters.

Authentication

Enter Google Cloud Project ID.

To connect to a specific Google service, you will either be using the OAuth2 authentication (default) or the authentication with a Google Service Account.  

Connection parameters

• Always stop after N minutes: the number of minutes to run the connector before stopping. The connector will stop even if there are new records in the queue.
• Max number of records to read: the total maximum number of records to read in one micro-batch. 
• Poll duration: how long (in milliseconds) the consumer should wait while fetching the data from the queue. 
• Number of retries before stop polling: the number of retries before stop polling if the poll returns no records. 
• Enable Message Ordering for Subscription: enable this property to create a subscription which will receive messages in the same order in which they were published. The messages must have an order key.
• Subscriber Preprocessor: a program in JavaScript that can be used to modify the source message or the destination name. Read more.
• Integration with CDC providers: the CDC provider. Select either Etlworks or Debezium if you are planning to use this connection for processing CDC events created by ETL CDC connectors or Debezium.
• Publish messages synchronously: enable this property to publish messages synchronously. Otherwise fire and forget.
• Add Order Key when publishing message: enable this property to add an order key when publishing messages. The order key will be set to topic name.
• Publisher Preprocessor: a program in JavaScript that can be used to modify the message which will be added to the topic. Read more. 

Subscriber preprocessor

Use Subscriber Preprocessor to modify the source message and/or destination name. 

Available scripting languages  

• JavaScript

Available variables

• event: the message serialized as JsonNode.
• fields: the source fields serialized as JsonNode.
• topic: the Google Pub/Sub topic.
• destination: the destination name.
• db - the source database name.
• schema - the source schema name.

Get the values of the column

var val = event.get('columns_name').asText();  

Modify the value of the column

Using this technique you can modify the values of the existing columns and add new columns to the stream.

event.put('columns_name', 'string');

Possible return values

• java.lang.Boolean.TRUE/ FALSE: if FALSE, the message will be skipped.
• TypedKeyValue<String, Boolean>: the key/value pair, where the key contains the destination name and the value is java.lang.Boolean.TRUE/FALSE.
• Anything else is ignored.

Set the destination name

By default, the destination table name or a filename is calculated using a template set in TO where any part of the TO can contain any of the following [tokens]:

• [table] - the source table name.
• [db] - the source database name.
• [schema] - the source schema.
• * - the source topic name.

The flow substitutes [tokens] on the values of the [tokens].

This example demonstrates how to configure a transformation to load data into the destination table in the public schema and with the same name as a source table: public.[table].

The developer can also set the destination name using JavaScript. Simply return (assign it to variable valuein the last line of the Consumer Preprocessor script) the instance of TypedKeyValue class, where the key = the new destination name.

Here is an example:

var topic.split('\\.', -1);
// assuming that the topic name includes database.schema.table and we only need table name
value = new TypedKeyValue('public.' + topic[2], java.lang.Boolean.TRUE);

Publisher Preprocessor

Use Publisher Preprocessor to modify the message and optionally order key which will be added to the topic.

Scripting languages  

• JavaScript

Variables

• message: the original message, serialized as String or ByteArrayStream.
• producerPackage: the instance of BaseProducerPackage.
• topic: the topic name.

Return values

• value = modified_message - the modified message
• value = null - the messages will not be added to the topic.
• TypedKeyValue<String, String>: the key/value pair, where the key contains the order key and the value is a message.