GCP Pub/Sub Source Connector #1224

davidsloan · 2024-05-17T13:55:36Z

This is the GCP Pub/Sub Source Connector, for copying data from GCP Pub/Sub into Kafka.

This supports the following configuration:

Connector Properties

Name	Description	Available Values
`connect.pubsub.gcp.auth.mode`	Authentication mode for connecting to GCP.	"`Credentials`", "`File`", or "`Default`"
`connect.pubsub.gcp.credentials`	For "`auth.mode`" credentials: GCP Authentication credentials string.
`connect.pubsub.gcp.file`	For "`auth.mode`" file: Local file path for file containing GCP authentication credentials.
`connect.pubsub.gcp.project.id`	GCP Project ID.
`connect.pubsub.output.mode`	Output Mode.	"`Compatibility`" (to match output of non-Lenses connector) or "`Default`".

KCQL Properties

Name	Description	Default Value
`batch.size`	Maximum size of batches to return per `poll` request (per PubSub partition).	1 000
`cache.ttl`	Time (millis) to retain unacknowledged messages stubs in the cache before releasing them.	1 hour
`queue.max`	Maximum number of messages to queue in memory (per PubSub partition) before bouncing them with a nack to be redelivered later	10 000

...ors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/util/ListSplitter.java

...kafka-connect-common/src/test/java/io/lenses/streamreactor/common/util/ListSplitterTest.java

...rc/main/java/io/lenses/streamreactor/connect/gcp/pubsub/source/GCPPubSubSourceConnector.java

...in/java/io/lenses/streamreactor/connect/gcp/pubsub/source/mapping/SourceRecordConverter.java

.../io/lenses/streamreactor/connect/gcp/pubsub/source/mapping/headers/MinimalHeaderMapping.java

...main/java/io/lenses/streamreactor/connect/gcp/pubsub/source/subscriber/PubSubSubscriber.java

GoMati-MU

Some remarks from me. I really like most of the logic around PubSub, but other classes (e.g. util ones) have more questionable approach.

...ka-connect-common/src/main/java/io/lenses/streamreactor/common/config/base/KcqlSettings.java

java-connectors/kafka-connect-query-language/src/main/java/io/lenses/kcql/Kcql.java

...ors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/util/ListSplitter.java

...nectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/util/MapUtils.java

...n/java/io/lenses/streamreactor/connect/gcp/pubsub/source/subscriber/LooselyBoundedQueue.java

...main/java/io/lenses/streamreactor/connect/gcp/pubsub/source/subscriber/PubSubSubscriber.java

...src/main/java/io/lenses/streamreactor/connect/gcp/pubsub/source/subscriber/SourceOffset.java

.../main/java/io/lenses/streamreactor/connect/gcp/pubsub/source/subscriber/SourcePartition.java

java-connectors/kafka-connect-gcp-pubsub/src/main/resources/gcp-pubsub-ascii-source.txt

...va/io/lenses/streamreactor/connect/gcp/pubsub/source/configdef/PubSubConfigSettingsTest.java

…treamreactor/common/config/base/KcqlSettings.java Co-authored-by: Mati Urban <157909548+GoMati-MU@users.noreply.github.com> Signed-off-by: David Sloan <33483659+davidsloan@users.noreply.github.com>

Changes addressed

davidsloan force-pushed the feat/gcp-pubsub branch 4 times, most recently from 6b14f2a to 55ee98e Compare May 17, 2024 15:22

davidsloan marked this pull request as ready for review May 17, 2024 15:31

davidsloan force-pushed the feat/gcp-pubsub branch 2 times, most recently from 9f68465 to 839f5bd Compare May 20, 2024 08:03