2022-09-26: This project has moved to a top-level repository at https://github.com/googleapis/java-pubsub-group-kafka-connector/ for future releases of the Pub/Sub Kafka connector. See #325 for the announcement.

The CloudPubSubConnector is a connector to be used with Kafka Connect to publish messages from Kafka to Google Cloud Pub/Sub or Pub/Sub Lite and vice versa.

CloudPubSubSinkConnector provides a sink connector to copy messages from Kafka to Google Cloud Pub/Sub. CloudPubSubSourceConnector provides a source connector to copy messages from Google Cloud Pub/Sub to Kafka. PubSubLiteSinkConnector provides a sink connector to copy messages from Kafka to Pub/Sub Lite. PubSubLiteSourceConnector provides a source connector to copy messages from Pub/Sub Lite to Kafka.

1. Regardless of whether you are running on Google Cloud Platform or not, you need to create a project and create service key that allows you access to the Cloud Pub/Sub API's and default quotas.

2. Create project on Google Cloud Platform. By default, this project will have multiple service accounts associated with it (see "IAM & Admin" within GCP console). Within this section, find the tab for "Service Accounts". Create a new service account and make sure to select "Furnish a new private key". Doing this will create the service account and download a private key file to your local machine.

3. Go to the "IAM" tab, find the service account you just created and click on the dropdown menu named "Role(s)". Under the "Pub/Sub" submenu, select "Pub/Sub Admin". Finally, the key file that was downloaded to your machine needs to be placed on the machine running the framework. An environment variable named GOOGLE_APPLICATION_CREDENTIALS must point to this file. (Tip: export this environment variable as part of your shell startup file).

   export GOOGLE_APPLICATION_CREDENTIALS=/path/to/key/file

You can download copy_tool.py, a single-file python script which downloads, sets up and runs the kafka connector in a single-machine configuration. This script requires:

1. python >= 3.5
2. requests installed
3. JAVA_HOME configured properly
4. GOOGLE_APPLICATION_CREDENTIALS set
5. A properties file with connector.class and other properties set

It can be invoked on mac/linux with:

python3 path/to/copy_tool.py --bootstrap_servers=MY_KAFKA_SERVER,OTHER_SERVER --connector_properties_file=path/to/connector.properties

or windows with:

python3 path\to\copy_tool.py --bootstrap_servers=MY_KAFKA_SERVER,OTHER_SERVER --connector_properties_file=path\to\connector.properties

A pre-built uber-jar is available for download with the latest release.

You can also build the connector from head, as described below.

1. Copy the pubsub-kafka-connector.jar to the place where you will run your Kafka connector.

2. Create a configuration file for the Pub/Sub connector and copy it to the place where you will run Kafka connect. The configuration should set up the proper Kafka topics, Pub/Sub topic, and Pub/Sub project. For Pub/Sub Lite, this should also set the correct location (google cloud zone). Sample configuration files for the source and sink connectors are provided at configs/.

3. Create an appropriate configuration for your Kafka connect instance. More information on the configuration for Kafka connect can be found in the Kafka Users Guide.

4. If running the Kafka Connector behind a proxy, you need to export the KAFKA_OPTS variable with options for connecting around the proxy. You can export this variable as part of a shell script in order ot make it easier. Here is an example:

export KAFKA_OPTS="-Dhttp.proxyHost=<host> -Dhttp.proxyPort=<port> -Dhttps.proxyHost=<host> -Dhttps.proxyPort=<port>"

In addition to the configs supplied by the Kafka Connect API, the Cloud Pub/Sub Connector supports the following configs:

In addition to the configs supplied by the Kafka Connect API, the Pub/Sub Lite Connector supports the following configs:

A pubsub message has two main parts: the message body and attributes. The message body is a ByteString object that translates well to and from the byte[] bodies of Kafka messages. For this reason, we recommend using a converter that produces primitive data types (i.e. integer, float, string, or bytes schema) where possible to prevent deserializing and reserializing the same message body.

Additionally, a Pubsub message size cannot exceed 10MB, so please check your broker's message.max.bytes configuration to prevent possible errors.

The sink connector handles the conversion in the following way:

• For integer, float, string, and bytes schemas, the bytes of the Kafka message's value are passed directly into the Pubsub message body.
• For map and struct types, the values are stored in attributes. Pubsub only supports string to string mapping in attributes. To make the connector as versatile as possible, the toString() method will be called on whatever object passed in as the key or value for a map and the value for a struct.
  • One additional feature is we allow a specification of a particular field or key to be placed in the Pubsub message body. To do so, set the messageBodyName configuration with the struct field or map key. This value is stored as a ByteString, and any integer, byte, float, or array type is included in the message body as if it were the sole value.
• For arrays, we only support primitive array types due to potential collisions of field names or keys of a struct/map array. The connector handles arrays in a fairly predictable fashion, each value is concatenated together into a ByteString object.
• In all cases, the Kafka key value is stored in the Pubsub message's attributes as a string, currently "key".

IMPORTANT NOTICE: There are three limitations to keep in mind when using Pubsub message attributes as stated on its documentation

• "Attributes per message: 100"
• "Attribute key size: 256 bytes"
• "Attribute value size: 1024 bytes"

If you enable copy of Kafka headers as Pubsub message attribute (it is disabled by default), the connector will copy only those headers meeting these limitations and will skip those that do not.

The source connector takes a similar approach in handling the conversion from a Pubsub message into a SourceRecord with a relevant Schema.

• The connector searches for the given kafka.key.attribute in the attributes of the Pubsub message. If found, this will be used as the Kafka key with a string schema type. Otherwise, it will be set to null.
• If the Pubsub message doesn't have any other attributes, the message body is stored as a byte[] for the Kafka message's value.
• However, if there are attributes beyond the Kafka key, the value is assigned a struct schema. Each key in the Pubsub message's attributes map becomes a field name, with the values set accordingly with string schemas. In this case, the Pubsub message body is identified by the field name set in "message", and has the schema types bytes.
  • In these cases, to carry forward the structure of data stored in attributes, we recommend using a converter that can represent a struct schema type in a useful way, e.g. JsonConverter.

Pub/Sub Lite's messages have the following structure:

class Message {
  ByteString key;
  ByteString data;
  ListMultimap<String, ByteString> attributes;
  Optional<Timestamp> eventTime;
}

This maps quite closely to the SinkRecord class, except for serialization. The table below shows how each field in SinkRecord will be mapped to the underlying message:

When a key, value or header value with a schema is encoded as a ByteString, the following logic will be used:

• null schemas are treated as Schema.STRING_SCHEMA
• Top level BYTES payloads are unmodified.
• Top level STRING payloads are encoded using copyFromUtf8.
• Top level Integral payloads are converted using copyFromUtf8(Long.toString(x.longValue()))
• Top level Floating point payloads are converted using copyFromUtf8(Double.toString(x.doubleValue()))
• All other payloads are encoded into a protobuf Value, then converted to a ByteString.
  • Nested STRING fields are encoded into a protobuf Value.
  • Nested BYTES fields are encoded to a protobuf Value holding the base64 encoded bytes.
  • Nested Numeric fields are encoded as a double into a protobuf Value.
  • Maps with Array, Map, or Struct keys are not supported.
    • BYTES keys in maps are base64 encoded.
    • Integral keys are converted using Long.toString(x.longValue())
    • Floating point keys are converted using Double.toString(x.doubleValue())

The source connector will perform a one to one mapping from SequencedMessage fields to their SourceRecord counterparts.

In addition, empty message.key fields will be converted to null and assigned round-robin to kafka partitions. Messages with identical, non-empty keys will be routed to the same kafka partition.

These instructions assume you are using Maven.

1. If you want to build the connector from head, clone the repository, ensuring to do so recursively to pick up submodules:

   git clone --recursive https://github.com/GoogleCloudPlatform/pubsub

   If you wish to build from a released version of the connector, download it from the Releases section in GitHub.

2. Unzip the source code if downloaded from the release version.

3. Go into the kafka-connector directory in the cloned repo or downloaded release.

4. Make the jar that contains the connector:

   mvn package

The resulting jar is at target/pubsub-kafka-connector.jar.