How to Make GCS Sink Connector Faster or Update Offfset

Confluent Cloud
1

Greetings,

Is there a way to reset the offsets of the Confluent Cloud plugin-version of the Google Cloud Storage Sink Connector? If not, what settings should I be using to ensure that the connector is getting through messages as quickly/efficiently as possible?

Situation:
I’ve got a connector that is using two topics that has been created via terraform. The first topic has ~70 million messages in it. The second topic has ~2 billion messages in it. The stakeholder who has requested this connector be set up would like for the latest messages to be processed/show up in the Google Cloud Storage bucket, but that hasn’t happened to this point and instead it seems like the connector is starting from the beginning of the messages that it has instead of the latest. As a result, the connector has been running for a day and a half and is showing that it is still ~1.9 billion messages behind.

What We’ve Tried:

  • One of the first things we did to lower the amount of messages in the queue was to decrease the retention time on the topics from 15 days to 7. This definitely lowered the number of messages behind by taking it down from ~7 billion to the ~2 billion it’s now at.
  • Set max tasks to 32.
  • Added the setting auto.offset.reset=latest and consumer.override.auto.offset.reset=latest in the config_nonsensitive section of the confluent_connector terraform resource.
  • Tried setting group.id and consumer.override.group.id in the config_nonsensitive section as well, but each time the connector is created it uses a new consumer group (the thought here was that we could create the connector/consumer group, delete the connector, update the offsets of the consumer group to latest, and then recreate the connector using that same consumer group).
  • Tried setting some of the record properties to different values to increase throughput. Current values:
Part Size in Multi-part Uploads: 5242880
Maximum span of record time (in ms) before rotation: 150000
Maximum span of record time (in ms) before scheduled rotation: 150000
Max poll records: 1000
Flush size: 1000
  • We also tried using kafka-consumer-groups.sh to update the offsets of the connector’s assigned consumer group both when it was running and when it was paused and it wouldn’t let us saying that it couldn’t do that while the consumer group was stable.
  • We tried using the REST API to stop the connector to hopefully stop the consumer group without deleting it, but we got 404 route_not_found when using the path https://api.confluent.cloud/connect/v1/environments/[environment id]/clusters/[cluster id]/connectors/[connector name]/stop.

Currently, the connector has been running for a day and a half, has processed only ~28 million messages, is only getting through a couple hundred messages a second, and isn’t keeping pace with the number of messages being produced to those two topics and the Messages behind queue continues to grow.

I’m still fairly new to Kafka, especially through Confluent Cloud, so any help getting the connector to use the latest offset or speeding the connector up would be greatly appreciated!