# Generalized Google Cloud Storage Source Connector for Confluent Platform The Generalized Kafka Connect Google Cloud Storage (GCS) Source connector can read data from any type of file naming convention listed under a GCS bucket, and the filenames don’t have to be in a specific format. As long as the files are in any of the supported formats (for example, JSON, Avro and Byte Array) the connector will be able to read them. #### IMPORTANT Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8) page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and are no longer available for download. ## Features The Generalized GCS Source connector includes the following features: - [At least once delivery](#generalized-gcs-source-at-least-once-delivery) - [Multiple tasks](#generalized-gcs-source-multiple-tasks) - [Client-side encryption](#generalized-gcs-source-csfle-sm) ### At least once delivery In the event of a task failure the connector guarantees no messages are lost, although the last few messages may be processed again. ### Multiple tasks The Generalized GCS Source connector supports running one or more tasks. You can specify the number of tasks in the `tasks.max` configuration parameter. This can lead to performance gains when multiple files need to be parsed. ### Client-side encryption This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html). ## Limitations The Generalized GCS Source connector has the following limitations: - The connector won’t reload data during the following scenarios: - Renaming a file which the connector has already read. - Uploading a newer version of a file with a new record. You should also be aware of the following connector actions: - The connector ignores any GCS object with a name that does not start with the configured `topics.dir` directory. This name is `topics/` by default. - The connector uses the connector name to store offsets on how much of the bucket it has processed. Deleting a connector and using the same name will not cause the connector to reprocess from the beginning but will save the progress of the original connector unless the corresponding entry in the offset topic is cleared. ## Install the Generalized GCS Source Connector You can install this connector by using the [confluent connect plugin install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command, or by manually downloading the ZIP file. ### Prerequisites - You must install the connector on every machine where Connect will run. - Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later. - Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later. - Java 1.8. - An installation of the latest (`latest`) connector version. To install the `latest` connector version, navigate to your Confluent Platform installation directory and run the following command: ```bash confluent connect plugin install confluentinc/kafka-connect-gcs-source:latest ``` You can install a specific version by replacing `latest` with a version number as shown in the following example: ```bash confluent connect plugin install confluentinc/kafka-connect-gcs-source:3.1.0 ``` #### Permissions The following permissions are required for the Generalized GCS Source connector: - `storage.buckets.get` - `storage.objects.get` - `storage.objects.list` Or, you may grant the Service Account the following roles on a bucket: - Storage Object Viewer - Storage Legacy Bucket Reader ### Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-gcs-source) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/install.html). ## License You can use this connector for a 30-day trial period without a license key. After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information. See [Confluent Platform license](../configuration_options.md#gcs-source-connector-license-config) for license properties and [Confluent License Properties](../configuration_options.md#gcs-src-license-topic-configuration) for information about the license topic. ## Configuration Properties For a complete list of configuration properties for this connector, see [Configuration Reference for Google Cloud Storage Source Connector for Confluent Platform](../configuration_options.md#gcs-source-config-options). For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see [Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster). ## Quick Start In the following example, the GCS Source connector reads all data listed under a specific GCS bucket and then loads them into a Kafka topic. It does not matter what file naming convention you use for writing data to the GCS bucket. 1. Upload the following data under a folder name `quickstart` within the targeted GCS bucket. In this example JSON format, which supports line-delimited JSON, concatenated JSON, and a JSON array of records, is used. ```json {"f1": "value1"} {"f1": "value2"} {"f1": "value3"} {"f1": "value4"} {"f1": "value5"} {"f1": "value6"} {"f1": "value7"} {"f1": "value8"} {"f1": "value9"} ``` 2. Install the connector through the [Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html) by running the following command from your Confluent Platform installation directory: ```bash confluent connect plugin install confluentinc/kafka-connect-gcs-source:latest ``` 3. Create a `quickstart-gcssource-generalized.properties` file with the following contents: ```json { "name": "quickstart-gcs-source", "config": { "connector.class": "io.confluent.connect.gcs.GcsSourceConnector", "tasks.max": "1", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "mode": "GENERIC", "topics.dir": "quickstart", "topic.regex.list": "quick-start-topic:.*", "format.class": "io.confluent.connect.gcs.format.json.JsonFormat", "gcs.bucket.name": "", "gcs.credentials.path": "", "value.converter.schemas.enable": "false" } } ``` 4. Load the Generalized GCS Source connector by running the following command: ```bash confluent local services connect connector load quickstart-gcs-source --config quickstart-gcssource-generalized.properties ``` 5. Verify the connector is in a `RUNNING` state: ```bash confluent local services connect connector status quickstart-gcs-source ``` 6. Verify the messages are being sent to Kafka: ```bash kafka-console-consumer \ --bootstrap-server localhost:9092 \ --topic quick-start-topic \ --from-beginning ``` 7. You should see output similar to the following: ```json {"f1": "value1"} {"f1": "value2"} {"f1": "value3"} {"f1": "value4"} {"f1": "value5"} {"f1": "value6"} {"f1": "value7"} {"f1": "value8"} {"f1": "value9"} ``` ### Property-based example 1. Create a `gcs-source-connector.properties` file with the following contents. This file is included with the connector in `etc/kafka-connect-gcs/gcs-source-connector.properties`. This configuration is used typically along with [standalone workers](/platform/current/connect/concepts.html#standalone-workers).: ```json { "name" : "GCSSourceConnector", "config" : { "format.class": "io.confluent.connect.gcs.format.avro.AvroFormat", "connector.class" : "io.confluent.connect.gcs.GcsSourceConnector", "gcs.bucket.name" : "", "gcs.credentials.path" : "", "tasks.max" : "1", "confluent.topic.bootstrap.servers" : "localhost:9092", "confluent.topic.replication.factor" : "1", "confluent.license" : "" } } ``` 2. Edit the `gcs-source-connector.properties` file to add the following properties: ```json { "transforms" : "AddPrefix", "transforms.AddPrefix.type" : "org.apache.kafka.connect.transforms.RegexRouter", "transforms.AddPrefix.regex" : ".*", "transforms.AddPrefix.replacement" : "copy_of_$0" } ``` Adding the previous properties renames the output topic of the messages to `copy_of_gcs_topic` which prevents a continuous feedback loop of messages. 3. Load the Generalized GCS Source connector. ```bash confluent local load gcs-source --config gcs-source-connector.properties ``` #### IMPORTANT Don’t use the local CLI commands in a production environment. The [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) is used in production environments. See [confluent local](https://docs.confluent.io/confluent-cli/current/command-reference/local/confluent_local_current.html#confluent-local-currentference/local/confluent_local_current.html#confluent-local-current). for more information about local CLI commands. 4. Verify the connector is in a `RUNNING` state. ```bash confluent local status gcs-source ``` 5. Verify messages are being sent to Kafka. ```bash kafka-avro-console-consumer \ --bootstrap-server localhost:9092 \ --property schema.registry.url=http://localhost:8081 \ --topic copy_of_gcs_topic \ --from-beginning | jq '.' ``` 6. The response should be 9 records as shown in the following example: ```bash {"f1": "value1"} {"f1": "value2"} {"f1": "value3"} {"f1": "value4"} {"f1": "value5"} {"f1": "value6"} {"f1": "value7"} {"f1": "value8"} {"f1": "value9"} ``` ### REST-based example 1. Use this setting with [distributed workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON, which can be used to read all the data list directly under a GCS bucket, to `config.json`, configure all of the required values, and use the following command to post the configuration to one of the distributed connect workers. Check here for more information about the Kafka Connect [REST API](/platform/current/connect/references/restapi.html). ```json { "name": "gcs-source-generalized", "config": { "connector.class": "io.confluent.connect.gcs.GcsSourceConnector", "tasks.max": "1", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "mode": "GENERIC", "topics.dir": " ", "topic.regex.list": "mytopic:.", "format.class": "io.confluent.connect.gcs.format.json.JsonFormat", "gcs.bucket.name": "", "gcs.credentials.path": "", "value.converter.schemas.enable": "false", "confluent.topic.bootstrap.servers" : "localhost:9092", "confluent.topic.replication.factor" : "1", "confluent.license" : " Omit to enable trial mode ", "transforms" : "AddPrefix", "transforms.AddPrefix.type" : "org.apache.kafka.connect.transforms.RegexRouter", "transforms.AddPrefix.regex" : ".", "transforms.AddPrefix.replacement" : "copy_of_$0" } } ``` #### NOTE Change the `confluent.topic.bootstrap.servers` property to include your broker address(es), and change the `confluent.topic.replication.factor` to 3 for staging or production use. 2. Use curl to post a configuration to one of the Kafka Connect Workers. Change `http://localhost:8083/` to the endpoint of one of your Kafka Connect worker(s). ```bash curl -s -X POST -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors ``` 3. Use the following command to update the configuration for an existing connector. ```bash curl -s -X PUT -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors/GCSSourceConnector/config ``` 4. To consume records written by the connector to the configured Kafka topic, run the following command: ```bash kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --topic copy_of_gcs_topic --from-beginning ``` ## Generalized GCS Source Connector Data Formats The Generalized GCS Source connector supports several data formats: - *Avro*: For supporting Avro format. To use Avro format, you must set the `format.class` property to `io.confluent.connect.gcs.format.avro.AvroFormat`. - *JSON*: For supporting JSON format. To use JSON format, you must set the `format.class` property to `io.confluent.connect.gcs.format.json.JsonFormat`. - *Raw Bytes*: For supporting Raw Bytes format. To use the Raw Bytes format, you must set the `format.class` property to `io.confluent.connect.gcs.format.bytearray.ByteArrayFormat`. ## Custom Credentials Provider You can configure the GCS connector to use a custom credentials provider, instead of the default one provided by the GCS connector. To do this, you implement a custom credentials provider, build it as a JAR file, and deploy the JAR file to use the custom provider. Complete the following steps to use a custom credentials provider: 1. **Set a custom credentials provider class**: Set the `gcs.credentials.provider.class` property to a class that implements the `com.google.api.gax.core.CredentialsProvider` interface. Configure the class to the fully qualified name of your custom credentials provider class. 2. **Configure additional settings (Optional)**: For additional configuration, prefix the configuration keys with `gcs.credentials.provider.` If your custom credentials provider needs to accept additional configuration, implement the `org.apache.kafka.common.Configurable` interface that lets the connector receive configurations that are prefixed with `gcs.credentials.provider.`. 3. **Ensure a public no-args constructor**: Your custom credentials provider class must have a public no-argument constructor. This is necessary because the connector creates an instance of the provider using this constructor. 4. **Package your provider**: Once your custom credentials provider class is implemented, package it into a JAR file. 5. **Copy the JAR file to Connect Worker**: Copy the built JAR file to the `share/java/kafka-connect-gcs` directory on all Connect workers. This step ensures that the GCS connector can access and use your custom credentials provider. ## Troubleshooting Connector and Task Failures ### Stack Trace You can use the Connect [REST API](/platform/current/connect/references/restapi.html) to check the status of the connectors and tasks. If a task or connector has failed, the `trace` field will include a reason and a stack trace.