Support writing general records to Pulsar sink #9590

sijie · 2021-02-15T23:42:51Z

No description provided.

sijie · 2021-02-16T00:42:46Z

The unit tests are added. Integration tests to be added.

codelipenghui · 2021-02-16T03:19:36Z

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

+            if (GenericRecord.class.isAssignableFrom(typeArg)) {
+                consumerConfig.setSchemaType(SchemaType.AUTO_CONSUME.toString());
+                SchemaType configuredSchemaType = SchemaType.valueOf(pulsarSinkConfig.getSchemaType());
+                if (SchemaType.AUTO_CONSUME != configuredSchemaType) {


Need to add consumerConfig.setSchemaType(pulsarSinkConfig.getSchemaType()); in this branch?

No. The schema type is already overwritten in line 419. This is just to log an info message to indicate that the schema type has been overwritten to AUTO_CONSUME.

eolivelli · 2021-02-16T11:29:12Z

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

@@ -400,7 +415,16 @@ public void close() throws Exception {
        ConsumerConfig consumerConfig = new ConsumerConfig();
        consumerConfig.setSchemaProperties(pulsarSinkConfig.getSchemaProperties());
        if (!StringUtils.isEmpty(pulsarSinkConfig.getSchemaType())) {
-            consumerConfig.setSchemaType(pulsarSinkConfig.getSchemaType());
+            if (GenericRecord.class.isAssignableFrom(typeArg)) {


@sijie initially you pointed out that working here in PulsarSink is not the right way, but we should only work on TopicSchema
#9481 (comment)

In fact I believe that in my PR #9481 I took the right way, driven by your suggestions.

I believe that this change is not enough in order to support by needs.

BTW if the integration test I added to #9481 works with this patch then we can converge to a good solution.
My goal is to get that usecase work, in the best way for the project for the mid/long term

@eolivelli Yes and no on my original comment.

My original comment is to make sure we returned the write schema information via TopicSchema. Because we are using AUTO_CONSUME in the PulsarSink to indicate GenericRecord are published to the Pulsar topic. AUTO_CONSUME can be used by both source and sink. In order to not impact sources, I didn't add the logic in TopicSchema. Instead, I add it in PulsarSink to make it more explicit, which results in one line of similar change as your initial change. But it doesn't your original and current implementation is in the right direction.

The main problem of your previous and current implementation on #9481 is you are trying to hijack the existing AVRO implementation to introduce the support of lazy schema initialization. The lazy schema initialization is already implemented as part of multi-schema write support. So you don't need to add such a hack.

eolivelli · 2021-02-16T22:58:45Z

@sijie

I totally agree that the main point here is to prevent the PulsarSink from creating the Producer and forcing a Schema on the topic in case of GenericRecord type.
So I am fine with this approach as well.

if you are okay I can merge this patch in my branch at #9481 (and revert the changes to TopicSchema) as we already have integration tests and I can continue the work.
But if you prefer I can close my PR and let you complete your patch, but please add an integration test like my one (that basically covers my usecase).

I just want to see this feature land to master branch and make it available to our users.

sijie · 2021-02-17T02:59:34Z

I will complete the integration tests here.

codelipenghui · 2021-02-23T04:05:32Z

/pulsarbot run-failure-checks

eolivelli

LGTM

Great work

Thanks @sijie

zymap · 2021-02-23T10:06:17Z

Failed tests:

~~~~~~~ SKIPPED -- [TestClass name=class org.apache.pulsar.tests.integration.io.GenericRecordSourceTest].testGenericRecordSource([])-------------- Starting test [TestClass name=class org.apache.pulsar.tests.integration.io.GenericRecordSourceTest].testGenericRecordSource([])-------
07:19:45.283 [TestNG-method=testGenericRecordSource-1:org.apache.pulsar.tests.integration.io.GenericRecordSourceTest@112] INFO  org.apache.pulsar.tests.integration.io.GenericRecordSourceTest - Run command : /pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource
07:19:45.288 [docker-java-stream-57079303:org.apache.pulsar.tests.integration.utils.DockerUtils$2@216] INFO  org.apache.pulsar.tests.integration.utils.DockerUtils - DOCKER.exec(ckmqvcaq-standalone:/pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource): Executing...
07:19:47.833 [docker-java-stream-57079303:org.apache.pulsar.tests.integration.utils.DockerUtils$2@221] INFO  org.apache.pulsar.tests.integration.utils.DockerUtils - DOCKER.exec(ckmqvcaq-standalone:/pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource): STDERR: Source class org.apache.pulsar.tests.integration.io.GenericRecordSource must be in class path
07:19:47.834 [docker-java-stream-57079303:org.apache.pulsar.tests.integration.utils.DockerUtils$2@221] INFO  org.apache.pulsar.tests.integration.utils.DockerUtils - DOCKER.exec(ckmqvcaq-standalone:/pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource): STDERR: Reason: Source class org.apache.pulsar.tests.integration.io.GenericRecordSource must be in class path
07:19:48.163 [docker-java-stream-57079303:org.apache.pulsar.tests.integration.utils.DockerUtils$2@236] INFO  org.apache.pulsar.tests.integration.utils.DockerUtils - DOCKER.exec(ckmqvcaq-standalone:/pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource): Done
07:19:48.167 [docker-java-stream-57079303:org.apache.pulsar.tests.integration.utils.DockerUtils$2@254] INFO  org.apache.pulsar.tests.integration.utils.DockerUtils - DOCKER.exec(ckmqvcaq-standalone:/pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource): completed with 1
07:19:48.167 [docker-java-stream-57079303:org.apache.pulsar.tests.integration.utils.DockerUtils$2@257] ERROR org.apache.pulsar.tests.integration.utils.DockerUtils - DOCKER.exec(ckmqvcaq-standalone:/pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource): completed with non zero return code: 1
stdout: 
stderr: Source class org.apache.pulsar.tests.integration.io.GenericRecordSource must be in class path

Reason: Source class org.apache.pulsar.tests.integration.io.GenericRecordSource must be in class path

!!!!!!!!! FAILURE-- [TestClass name=class org.apache.pulsar.tests.integration.io.GenericRecordSourceTest].testGenericRecordSource([])-------
Error:  Tests run: 5, Failures: 1, Errors: 0, Skipped: 1, Time elapsed: 206.535 s <<< FAILURE! - in TestSuite
Error:  pulsar-standalone-suite(org.apache.pulsar.tests.integration.io.GenericRecordSourceTest)  Time elapsed: 2.904 s  <<< FAILURE!
org.apache.pulsar.tests.integration.docker.ContainerExecException: /pulsar/bin/pulsar-admin sources create --name test-state-source-isolqgjy --destinationTopicName test-state-source-output-zrbylspk --archive /pulsar/examples/java-test-functions.jar --classname org.apache.pulsar.tests.integration.io.GenericRecordSource failed on 267d5fc4c5f0ad1512ae6ba8588f32236a6d72ddd840ba940a24d9d2a94de872 with error code 1
	at org.apache.pulsar.tests.integration.utils.DockerUtils$2.onComplete(DockerUtils.java:259)
	at org.testcontainers.shaded.com.github.dockerjava.core.exec.AbstrAsyncDockerCmdExec$1.onComplete(AbstrAsyncDockerCmdExec.java:51)
	at org.testcontainers.shaded.com.github.dockerjava.core.DefaultInvocationBuilder.lambda$executeAndStream$1(DefaultInvocationBuilder.java:276)
	at java.lang.Thread.run(Thread.java:748)

[INFO] 
[INFO] Results:
[INFO] 
Error:  Failures: 
Error:  org.apache.pulsar.tests.integration.io.GenericRecordSourceTest.pulsar-standalone-suite(org.apache.pulsar.tests.integration.io.GenericRecordSourceTest)
[INFO]   Run 1: PASS
Error:    Run 2: GenericRecordSourceTest.testGenericRecordSource » ContainerExec /pulsar/bin/pu...
[INFO] 
[INFO] 
Error:  Tests run: 4, Failures: 1, Errors: 0, Skipped: 0

eolivelli · 2021-03-09T11:12:29Z

@sijie do you have time to complete this patch.

It is very useful for a couple of usecases I saw and I really would like this work to land to master
the patch has already been approved by @codelipenghui and @freeznet

freeznet · 2021-03-10T02:35:35Z

@sijie do you have time to complete this patch.

It is very useful for a couple of usecases I saw and I really would like this work to land to master
the patch has already been approved by @codelipenghui and @freeznet

@eolivelli i am work with @sijie to fix the ci failed issue, will update and finish this pr soon.

…_general_record

eolivelli

Thanks you guys for this work.
It is really a good step forward.

But I need a little more, please check my latest comment

eolivelli · 2021-03-14T06:21:48Z

...test-functions/src/main/java/org/apache/pulsar/tests/integration/io/GenericRecordSource.java

+        Thread.sleep(20);
+
+        int value = count.incrementAndGet();
+        GenericRecord record = schema.newRecordBuilder()


@sijie @codelipenghui unfortunately this is not exactly like my original test case, that reproduced my use case.
That is to be able to push an object that implements GenericRecord.

Here you are using the builder provided by Pulsar but this is mo enough for me, because my user would like to use an object from his own domain, just by implementing Pulsar GenericRecord java interface, because we will save resources (allocations and cycles)

https://github.com/apache/pulsar/pull/9481/files#diff-bbdf586ddad181a0a9dae17974b19ca5cbbce398716ec7fa5b4c45b69be58f41R66

GenericRecord was designed to be from RecordSchemaBuilder. It doesn't expect people to implement the GenericRecord directly. I don't understand the allocations and cycles issue. If it is an allocations or cycles issue, it should be fixed in RecordSchemaBuilder. It shouldn't be done by just implementing GenericRecord.

@sijie okay
I will check downstream if using RecordSchemaBuilder is a valid option.
thanks

(cherry picked from commit b826e03)

Support writing general records to Pulsar sink

f8b2d1d

sijie added area/function area/connector labels Feb 15, 2021

sijie added this to the 2.8.0 milestone Feb 15, 2021

sijie self-assigned this Feb 15, 2021

sijie mentioned this pull request Feb 15, 2021

Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO #9481

Closed

Add unit tests

cd8e6fc

sijie changed the title ~~[WIP] Support writing general records to Pulsar sink~~ Support writing general records to Pulsar sink Feb 16, 2021

codelipenghui approved these changes Feb 16, 2021

View reviewed changes

codelipenghui requested review from wolfstudy, 315157973, hangc0276, jiazhai, merlimat and zymap February 16, 2021 02:09

codelipenghui reviewed Feb 16, 2021

View reviewed changes

eolivelli requested changes Feb 16, 2021

View reviewed changes

Add integration tests

acd51c7

freeznet approved these changes Feb 23, 2021

View reviewed changes

eolivelli approved these changes Feb 23, 2021

View reviewed changes

Merge remote-tracking branch 'apache/master' into write_general_record

27613ee

eolivelli mentioned this pull request Mar 12, 2021

Pulsar function Should support Generic output Message #9892

Closed

Merge remote-tracking branch 'apache/master' into write_general_record

0c90e99

freeznet and others added 2 commits March 12, 2021 19:23

fix ci

6bb2137

Merge remote-tracking branch 'origin/write_general_record' into write…

1c4a03b

…_general_record

codelipenghui merged commit b826e03 into apache:master Mar 14, 2021

eolivelli reviewed Mar 14, 2021

View reviewed changes

fmiguelez pushed a commit to fmiguelez/pulsar that referenced this pull request Mar 16, 2021

Support writing general records to Pulsar sink (apache#9590)

03b3a26

eolivelli pushed a commit to datastax/pulsar that referenced this pull request May 26, 2021

Support writing general records to Pulsar sink (apache#9590)

305db0e

(cherry picked from commit b826e03)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support writing general records to Pulsar sink #9590

Support writing general records to Pulsar sink #9590

sijie commented Feb 15, 2021

sijie commented Feb 16, 2021

codelipenghui Feb 16, 2021

sijie Feb 16, 2021

eolivelli Feb 16, 2021

sijie Feb 16, 2021

eolivelli commented Feb 16, 2021

sijie commented Feb 17, 2021

codelipenghui commented Feb 23, 2021

eolivelli left a comment

zymap commented Feb 23, 2021

eolivelli commented Mar 9, 2021

freeznet commented Mar 10, 2021

eolivelli left a comment

eolivelli Mar 14, 2021

sijie Mar 16, 2021

eolivelli Mar 17, 2021

Support writing general records to Pulsar sink #9590

Support writing general records to Pulsar sink #9590

Conversation

sijie commented Feb 15, 2021

sijie commented Feb 16, 2021

codelipenghui Feb 16, 2021

Choose a reason for hiding this comment

sijie Feb 16, 2021

Choose a reason for hiding this comment

eolivelli Feb 16, 2021

Choose a reason for hiding this comment

sijie Feb 16, 2021

Choose a reason for hiding this comment

eolivelli commented Feb 16, 2021

sijie commented Feb 17, 2021

codelipenghui commented Feb 23, 2021

eolivelli left a comment

Choose a reason for hiding this comment

zymap commented Feb 23, 2021

eolivelli commented Mar 9, 2021

freeznet commented Mar 10, 2021

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli Mar 14, 2021

Choose a reason for hiding this comment

sijie Mar 16, 2021

Choose a reason for hiding this comment

eolivelli Mar 17, 2021

Choose a reason for hiding this comment