Confluent Cloud Plugin
Introduction
Confluent Cloud is a streaming data service with Apache Kafka delivered as a managed service.
User Storie(s)
As a pipeline developer, I would like to stream data real time from Confluent cloud based on the specified schema by the user
As a pipeline developer, I would like to stream data real time to Confluent Cloud and perform serialization during event streaming
As a pipeline developer I would like to capture records that did not get delivered downstream to Confluent Cloud for analysis
Plugin Type
Prerequisites
Kafka Broker: Confluent Platform 3.3.0 or above, or Kafka 0.11.0 or above
Connect: Confluent Platform 3.3.0 or above, or Kafka 0.11.0 or above
Java 1.8
Properties
Real time source
Property | description | Mandatory or Optional |
|---|---|---|
referenceName | Uniquely identify the source | Yes |
Kafka cluster credential API key | API key in order to connect to the Confluent cluster | Yes |
Kafka cluster credential secret | Secret in order to connect to the Confluent cluster | Yes |
Kafka Zookeeper | The connect string location of ZooKeeper. Either that or the list of brokers is required | Required if brokers not specified |
Kafka brokers | Comma-separated list of Kafka brokers. Either that or the ZooKeeper quorum is required | Required if zookeeper not specified |
Kafka partition | Number of partitions | Yes |
Kafka offset | The initial offset for the partition |
|
Kafka topic | List of topics which we are listening to for streaming | Yes |
Schema registry URL | URL endpoint for the schema registry on Confluent Cloud or self hosted schema registry URL | No |
Schema registry API key | API key | No |
Schema registry secret | Secret | No |
Format | Specify the format for the Kafka event. Any supported format by CDAP is supported. Default output is key and Value as bytes | No |
Real time sink
Property | description | Type | Mandatory |
|---|---|---|---|
Reference Name | Uniquely identify the sink | String | Yes |
Kafka cluster credential API key | API key in order to connect to the Confluent cluster | String | Yes |
Kafka cluster credential secret | Secret in order to connect to the Confluent cluster | String | Yes |
Kafka brokers | Comma-separated list of Kafka brokers | String | Yes |
Async | Specifies whether writing the events to broker is Asynchronous or Synchronous | Select | Yes |
partitionfield | Specifies the input fields that need to be used to determine the partition id | Int or Long | Yes |
key | Specifies the input field that should be used as the key for the event published into Kafka. | String | Yes |
Kafka topic | List of topics to which the data should be published to | String | Yes |
format | Specifies the format of the event published to Confluent cloud | String | Yes |
References
https://docs.confluent.io/current/connect/kafka-connect-bigquery/index.html
https://docs.confluent.io/current/cloud/connectors/cc-gcs-sink-connector.html#cc-gcs-connect-sink
https://docs.confluent.io/current/quickstart/cloud-quickstart/index.html
https://docs.confluent.io/current/cloud/limits.html#cloud-limits
Design / Implementation Tips
Design
Approach(s)
Properties
Security
Limitation(s)
Future Work
Some future work – HYDRATOR-99999
Another future work – HYDRATOR-99999
Test Case(s)
Test case #1
Test case #2
Sample Pipeline
Please attach one or more sample pipeline(s) and associated data.
Pipeline #1
Pipeline #2
Table of Contents
- 1 Introduction
- 2 User Storie(s)
- 3 Plugin Type
- 4 Prerequisites
- 5 Properties
- 6 Design
- 6.1 Approach(s)
- 6.2 Properties
- 6.3 Security
- 6.4 Limitation(s)
- 6.5 Future Work
- 7 Test Case(s)
- 8 Sample Pipeline
- 8.1 Pipeline #1
- 8.2 Pipeline #2
Checklist
Comments
@Bhooshan Mogal @Vasudha Gupta (Unlicensed) Could you help me with several questions about this plugin?
There are existing Kafka plugins in kafka-plugins repo but it seems there are several problems with them: 1) they support Kafka 0.8 and 0.10, both are deprecated. 2) they export same packages in bundles. Should we make any changes to existing plugins? Should I implement Confluent Cloud plugins in kafka-plugins repo?
As I can see, the only difference from Kafka is Confluent Schema Registry. Could you please confirm that I only have to add schema registry url parameter to what is available in Kafka plugins.
@Mikkin Patel (Unlicensed) any thoughts on how to go about this?
Discussed with @Albert Shau offline as well about this. @Bhooshan Mogal since Kafka 0.8 and 0.10 are deprecated, what is the recommended path forward for these plugins?
@Maksym Lozbin to answer your questions:
We can add the Confluent cloud plugin as a separate module in the same repo.
We should add support for Schema Registry if we are sourcing data from Confluent Kafka. The user should be free to point to Confluent's Schema Registry or their self hosted Schema Registry.
Additionally, @Bhooshan Mogal can you please help complete the user story? Would be good to clarify use case differences when they are sending data to/from Confluent Kafka.
I am working on an updated version of the details for Confluent plugin and should have it done by EOD today. @Bhooshan Mogal you probably still need to make the decision regarding older plugins and the path there
From a product perspective, as a user, I want to be able to use my data in Confluent Cloud/Kafka in CDAP pipelines.
If that means we need to update the existing Kafka plugins, then yes we should do it.
A few requirements:
Confluent Cloud plugins need to show up in the UI differently from Kafka plugins
If Kafka plugins need to be updated to get Confluent Cloud to work, we should do it, without actually removing support for 0.8 and 0.10 right now. Perhaps create a newer version of Kafka plugins first.
In the Confluent Cloud plugins, support schema registry and Confluent's authentication mechanism.
For the old Kafka plugin, we should use the same style of deprecation (add a note in the name and description of the plugin saying its deprecated). However, we cannot remove the plugins, until Kafka actually stops support for these versions.
Does that help clarify @Maksym Lozbin @Mikkin Patel @Vasudha Gupta (Unlicensed)
@Maksym Lozbin based on my research, your assessment is accurate. Schema registry URL is the main addition in case we are sourcing data from Confluent Cloud. In addition to that cluster credential details including API key and Secret will have to be specified to connect to the cluster itself. Also based on my research we have to use an update version of Kafka, older versions wont suffice. Most of the user fields can be leveraged as available in our Kafka plugins (assuming Kafka updated version is used) along with the net new additions of cluster connection details as well as schema URL/custom schema URL specified by the user
For the sink, there will be no schema registry URL since its just a stream to a Kafka topic - hence I kept the same fields as in the already available plugins. Here also the cluster credentials details fields would need to be added.
@Bhooshan Mogalplease take a look and let me know if you disagree on the decision for upgrade to newer Kafka version
@Vasudha Gupta (Unlicensed) thanks for the research. But I'm not sure that sink plugin should not have schema URL. How user will later read data from Kafka without having schema for stored events? Or do you mean we should use another serializer?
I agree, we need Schema Registry support on both source and sink. We can create a new Kafka plugin module that can take in Schema Registry URL, optionally, if the user has one. Separately, we need to work on deprecating the older ones as @Bhooshan Mogal mentioned above.
@Bhooshan Mogal could you please clarify what is finalized approach we have to follow.
cc @Albert Shau.
The decisions are in the comment above - Re: Confluent Cloud Plugin
@Maksym Lozbin Makes sense. I was thinking more on the serialization route as you mentioned but I think you are right in order to subsequently read it downstream, the schema would be needed. I will add those columns in
@Bhooshan Mogal @Vasudha Gupta (Unlicensed) Please let us know if there are expected any details to this document.
Also we see it's selected batch sink, but we originally thought that it has to be Confluent Cloud realtime Source and Sink plugins.