Add ability to use Kafka's sinks as pulsar sinks #9927

dlg99 · 2021-03-16T18:55:55Z

Motivation

Provide a way to use Kafka-Connect Sink as a Pulsar Sink, in cases like:

company has custom kafka sink and want to try the pulsar out
no corresponding pulsar sink exists
etc.

Modifications

Added KafkaConnectSink. Kafka Schema is autodetected from key/value itself for primitive values with option to unwrap KeyValue. GenericRecord support is TBD, depends on the changes Enrico is working on.
Added profile to pulsar-io/kafka-connect-adaptor to build nar with kafka-connect connector included

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

Added unit tests.
Tested locally as

Ran pulsar standalone bin/pulsar standalone
add dependency

       <dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>connect-file</artifactId>
          <version>${kafka-client.version}</version>
        </dependency>

into pulsar-io/kafka-connect-adaptor-nar/pom.xml
Built nar as `mvn -f pulsar-io/kafka-connect-adaptor-nar/pom.xml clean package -DskipTests to include kafka's connect-file sink into the nar.
Ran test nar as

bin/pulsar-admin sinks localrun -a ./pulsar-io/kafka-connect-adaptor-nar/target/pulsar-io-kafka-connect-adaptor-nar-2.8.0-SNAPSHOT.nar --name kwrap --namespace public/default/ktest --parallelism 1 -i my-topic --sink-config-file ~/sink.yaml

with

$ cat ~/sink.yaml
processingGuarantees: "EFFECTIVELY_ONCE"
configs:
  "topic": "my-topic"
  "offsetStorageTopic": "kafka-connect-sink-offset"
  "pulsarServiceUrl": "pulsar://localhost:6650/" 
  "kafkaConnectorSinkClass": "org.apache.kafka.connect.file.FileStreamSinkConnector"
  "kafkaConnectorConfigProperties":
    "file": "/tmp/sink_test.out"

set topic schema as

bin/pulsar-admin schemas upload --filename ~/schema.json my-topic

$ cat ~/schema.json
{
    "type": "STRING",
    "schema": "",
    "properties": {
        "key1": "value1"
    }
}

message produced as

bin/pulsar-client produce my-topic --messages "hello-pulsar"

and got

$ cat /tmp/sink_test.out
hello-pulsar

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (not documented yet)
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation
TBD following review

dlg99 · 2021-03-16T18:57:28Z

this replaces #9825 - reworked according to @sijie 's feedback
@sijie and @eolivelli please take a look.

yuvalgut · 2021-03-16T19:16:59Z

cool! that will be really useful.
for supporting avro we can check if the Object is type of org.apache.pulsar.client.impl.schema.generic.GenericAvroRecord and then get hold of the actual avro record with '.getAvroRecord()'.
io.confluent.connect.avro.AvroData will let us to convert the avro into kafka SchemaAndValue:
avroData.toConnectData(avroRecord.getSchema(), avroRecord) and finally generate kafka SinkRecord.

just throwing some ideas...

dlg99 · 2021-03-16T19:22:05Z

@yuvalgut thank you!
I postponed dealing with GenericRecords for now to make use of the changes @eolivelli is working on, i.e. #9844

eolivelli

Awesome work.

I am not much convinced about the final packaging, because if you want to run an existing Kafka Sink you have to build this module and add that Sink as dependency (this is the same work we are doing with Debezium source?).
I wonder if we can (as a follow up work) add the a ability to pass an additional file (the jar for the Sink implementation) as configuration and dynamically load it.
Probably the new package management facility (still under development in current master branch) will help?

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaSinkTaskContext.java

dlg99 · 2021-03-16T22:53:54Z

I am not much convinced about the final packaging, because if you want to run an existing Kafka Sink you have to build this module and add that Sink as dependency (this is the same work we are doing with Debezium source?).

Final packaging is tricky.
What I have is an option to get started quickly.
I cannot include all possible kafka connect modules - most of the people will need only small subset of them (that's even if this won't create dependency conflicts), others will need some custom connector not available publicly.
Passing a jar assumes the fat jar -> either shading (size of nar increases) or dependency conflicts.
Merging two nars is an option https://jslabonte.wordpress.com/2012/02/01/how-to-merge-nar-files/
Either way, another profile is not for production use (I can't imagine file sink being very useful) but to provide an example.

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

sijie · 2021-03-17T05:02:18Z

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

+        final int partition = 0;
+        final Object key;
+        final Object value;
+        if (unwrapKeyValueIfAvailable && sourceRecord.getValue() instanceof KeyValue) {


You might also need to another branch to check if Record is a KVRecord.

@sijie KVRecord is not handled by Sinks, it is used only for Sources.
see #10113

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

sijie · 2021-03-17T05:28:05Z

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaSinkTaskContext.java

+    }
+
+    @Override
+    public void offset(TopicPartition topicPartition, long l) {


same comment above.

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaSinkTaskContext.java

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

sijie · 2021-03-17T05:43:20Z

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

+            sourceRecord.fail();
+            return;
+        }
+        pendingFlush.whenComplete((ignore, ex) -> {


I also don't think you need to register the callback here. Instead, I think you need to track the pulsar source records in the context in order to make sure the "offsets" are correctly recorded. once you call kafka sink to flush, you need to call sink context to flush. If you successfully flush, you can call sourceRecord to ack, otherwise you fail the source records.

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

dlg99 · 2021-03-24T00:44:38Z

updated to reflect CR comments (except some that affect pulsar-io/context and can be done separately/later)
rebased on top of #9947

### Motivation Looking at #9927 (comment) there is no way to reliably get partition number. There is Optional<String> getPartitionId() which returns: - "{topic}-{partition}" in some Record implementation - String(partition) in other implementation (parsing back integer is a waste) - empty Optional (ok) in others ### Modifications Added `default Optional<Integer> getPartitionIndex()` to the `Record` interface. Return partition number where appropriate.

dlg99 · 2021-04-19T17:24:24Z

rebased on latest master

eolivelli

I did my final pass of review.
Overall is good, I left some final comments

I hope we can merge this patch soon, this way we can continue to build up more features over this bridge to Kafka Connect.

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

...t-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaConnectSinkConfig.java

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaSinkTaskContext.java

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/GenericObjectWrapper.java

...ect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/SchemaedFileStreamSinkTask.java

eolivelli

Looks great to me

@sijie can you please review again ?

I believe we can merge this patch and let @dlg99 follow up with other patches.

eolivelli · 2021-04-27T15:25:23Z

@sijie can you please review ? this patch has been hanging for many days.

@merlimat @rdhabalia @codelipenghui can you help with merging this patch ?

…10446) Fixes #10445 ### Motivation SinkContext should expose Subscription type to the Sink More context: #9927 (comment) Needed for #9927 ### Modifications Added `getSubscriptionType()` to the `SinkContext` interface and `ContextImpl`

…pache#10446) Fixes apache#10445 ### Motivation SinkContext should expose Subscription type to the Sink More context: apache#9927 (comment) Needed for apache#9927 ### Modifications Added `getSubscriptionType()` to the `SinkContext` interface and `ContextImpl`

dlg99 · 2021-05-18T21:12:41Z

@sijie please take another look

sijie · 2021-06-01T16:21:25Z

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

+    }
+
+    private void flushIfNeeded(boolean force) {
+        if (force || currentBatchSize.get() >= maxBatchSize) {


@dlg99 This will cause submitting multiple flush tasks, no? Because currentBatchSize will always be larger than maxBatchSize before the flush action is taken.

This potentially will enqueue a lot of tasks in the flush queue which will burn CPU.

sijie · 2021-06-01T16:26:52Z

...t-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaConnectSinkConfig.java

+            required = true,
+            defaultValue = "",
+            help = "Pulsar service URL to use for the offset store.")
+    private String pulsarServiceUrl;


This field will potentially be removed after PulsarClient is exposed to the context. So you might want to document it.

* Add ability to use Kafka's sinks as Pulsar sinks * Get topic/partition from the record * Making use of GenericObject * Added SubscriptionType check and tests * support for topic/partition pause/resume/seek * added comment; only one flush in progress is allowed

dlg99 mentioned this pull request Mar 16, 2021

[pulsar-io] Add ability to use Kafka's sinks as pulsar sinks #9825

Closed

1 task

eolivelli reviewed Mar 16, 2021

View reviewed changes

...kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java Outdated Show resolved Hide resolved

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarKafkaSinkTaskContext.java Outdated Show resolved Hide resolved

sijie assigned dlg99 Mar 17, 2021

sijie added this to the 2.8.0 milestone Mar 17, 2021

sijie added the area/connector label Mar 17, 2021

sijie requested changes Mar 17, 2021

View reviewed changes

dlg99 mentioned this pull request Mar 17, 2021

add getPartitionIndex() to the Record<> #9947

Merged

1 task

dlg99 force-pushed the kafka-connect-adaptor-sink branch from 5046934 to 0f64b14 Compare March 24, 2021 00:10

dlg99 requested a review from sijie March 24, 2021 20:54

dlg99 force-pushed the kafka-connect-adaptor-sink branch 2 times, most recently from b5ca187 to 0b44bbd Compare April 1, 2021 20:07

dlg99 force-pushed the kafka-connect-adaptor-sink branch 3 times, most recently from 4beb265 to 7f08a23 Compare April 19, 2021 17:13

eolivelli requested changes Apr 20, 2021

View reviewed changes

dlg99 force-pushed the kafka-connect-adaptor-sink branch from 7f08a23 to 60aa808 Compare April 21, 2021 00:49

dlg99 requested a review from eolivelli April 21, 2021 15:26

eolivelli approved these changes Apr 22, 2021

View reviewed changes

codelipenghui requested review from merlimat, codelipenghui, congbobo184 and gaoran10 April 23, 2021 00:49

dlg99 mentioned this pull request Apr 29, 2021

[Issue 10445][pulsar-io] Exposed SubscriptionType in the SinkContext #10446

Merged

1 task

sijie mentioned this pull request Apr 30, 2021

ISSUE-10445: SinkContext should expose Subscription type to the Sink streamnative/pulsar-archived#2472

Closed

dlg99 force-pushed the kafka-connect-adaptor-sink branch from 94d18f6 to f7232b4 Compare May 4, 2021 23:41

dlg99 mentioned this pull request May 7, 2021

SinkContext: ability to seek/pause/resume consumer for a topic #10498

Merged

1 task

dlg99 force-pushed the kafka-connect-adaptor-sink branch from f7232b4 to c080b7f Compare May 10, 2021 20:57

dlg99 added 10 commits May 18, 2021 10:46

Add ability to use Kafka's sinks as pulsar sinks

f4d325e

CR feedback

f111ab5

Get topic/partiton from the record

6b57753

Making use of GenericObject

d9284f1

CR feedback

48f8875

CR feedback

341419a

CR feedback - in progress

0c05dbd

CR feedback. Still ToDo: handle subscriptionType check, PR is pending

5f4d1c7

Added SubscriptionType check and tests

5e3e4ae

support for topic/partition pause/resume/seek

c1e7858

dlg99 force-pushed the kafka-connect-adaptor-sink branch from c080b7f to c1e7858 Compare May 18, 2021 20:53

dlg99 requested a review from sijie May 18, 2021 21:12

codelipenghui modified the milestones: 2.8.0, 2.9.0 May 26, 2021

sijie reviewed Jun 1, 2021

View reviewed changes

CR feedback: added comment; only one flush in progress is allowed

b5593c1

sijie approved these changes Jun 1, 2021

View reviewed changes

lhotari approved these changes Jun 2, 2021

View reviewed changes

lhotari merged commit a2604c8 into apache:master Jun 2, 2021

dlg99 deleted the kafka-connect-adaptor-sink branch October 14, 2021 23:29

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add ability to use Kafka's sinks as pulsar sinks #9927

Add ability to use Kafka's sinks as pulsar sinks #9927

dlg99 commented Mar 16, 2021 •

edited

Loading

dlg99 commented Mar 16, 2021

yuvalgut commented Mar 16, 2021

dlg99 commented Mar 16, 2021

eolivelli left a comment

dlg99 commented Mar 16, 2021

sijie Mar 17, 2021

eolivelli Apr 21, 2021

sijie Mar 17, 2021

sijie Mar 17, 2021

dlg99 commented Mar 24, 2021

dlg99 commented Apr 19, 2021

eolivelli left a comment

eolivelli left a comment

eolivelli commented Apr 27, 2021

dlg99 commented May 18, 2021

sijie Jun 1, 2021

sijie Jun 1, 2021

sijie Jun 1, 2021

Add ability to use Kafka's sinks as pulsar sinks #9927

Add ability to use Kafka's sinks as pulsar sinks #9927

Conversation

dlg99 commented Mar 16, 2021 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

dlg99 commented Mar 16, 2021

yuvalgut commented Mar 16, 2021

dlg99 commented Mar 16, 2021

eolivelli left a comment

Choose a reason for hiding this comment

dlg99 commented Mar 16, 2021

sijie Mar 17, 2021

Choose a reason for hiding this comment

eolivelli Apr 21, 2021

Choose a reason for hiding this comment

sijie Mar 17, 2021

Choose a reason for hiding this comment

sijie Mar 17, 2021

Choose a reason for hiding this comment

dlg99 commented Mar 24, 2021

dlg99 commented Apr 19, 2021

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli commented Apr 27, 2021

dlg99 commented May 18, 2021

sijie Jun 1, 2021

Choose a reason for hiding this comment

sijie Jun 1, 2021

Choose a reason for hiding this comment

sijie Jun 1, 2021

Choose a reason for hiding this comment

dlg99 commented Mar 16, 2021 •

edited

Loading