[PIP-193] [feature][connectors] Add support for a transform Function in Sinks #16740

cbornet · 2022-07-22T10:07:21Z

Motivation

For PIP 193 : Sink preprocessing Function (#16739).
Implements changes to support the Sink transform function except the LocalRunner support.

Modifications

See #16739 Implementation section

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

TODO

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

The public API: yes
The rest endpoints: yes
The admin cli options: yes

Documentation

Check the box below or label this PR directly.

Need to update docs?

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

eolivelli

LGTM

shibd · 2022-07-27T13:35:11Z

...ctions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

@@ -384,7 +386,12 @@ private void sendOutputMessage(Record srcRecord, Object output) throws Exception
        try {
            this.sink.write(sinkRecord);
        } catch (Exception e) {
-            log.info("Encountered exception in sink write: ", e);
+            if (e instanceof ClassCastException && functionClassLoader != componentClassLoader) {


Can we do this check at sink create or start?

you can't because each record can be different

in Java even if you declare a class with some Generic types, you can always force the system to return something wrong.
so this check must stay here.

There is also a question about the type of input for user-defined functions. In the current implementation, the consumer in the sink sets the schema according to the generic definition of the sink. usually, we use GenericObject, If the user-defined function does not match the generic, the current logic will go to the following logic instead of being sent to the sink, is this the expected behavior?

pulsar/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

Lines 354 to 360 in 5df15dd

if (result.getUserException() != null) {

Exception t = result.getUserException();

log.warn("Encountered exception when processing message {}",

srcRecord, t);

stats.incrUserExceptions(t);

srcRecord.fail();

} else {

Maybe we need to validate the input parameter types of the user-defined function at the sink start?

If you have a Function with the Sink we use the input type of the Function to configure the PulsarSource instead of the Sink type. See here.
Does it answer your concerns ?

Oh, sorry. I miss this.

However, if the user wants to use it, the input parameter type may need to be carefully defined. In general, users can only use GenericObject, which triggers the registration of auto_consumer type schema, because only then can multiple topic inputs be processed.

Am I right?

You can use the same input type as if you had a regular Sink or Function. If you want to consume any type of Schema then yes, you should use GenericObject or byte[]. If you know that the Schema will always be String, you can use String as input.

...ons/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RecordFunction.java

eolivelli · 2022-08-24T07:59:22Z

...ctions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

        }
        AbstractSinkRecord<?> sinkRecord;
        if (output instanceof Record) {
-            sinkRecord = new OutputRecordSinkRecord<>(srcRecord, (Record) output);
+            Record record = (Record) output;


what about moving this block to a method ?

eolivelli · 2022-08-24T08:00:42Z

...ctions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

+                    finalSchema = schema;
+                }
+
+                sinkRecord = new OutputRecordSinkRecord<>(srcRecord, record) {


what about creating a named class ? it will help Sink developers when they dump the "Record" object (and especially us when we will debug problems about this feature :-) )

I changed the code to use distinct constructors. It does not introduce a new class though.

eolivelli · 2022-08-24T08:01:19Z

...ctions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

@@ -881,4 +963,98 @@ private void setupOutput(ContextImpl contextImpl) throws Exception {
            Thread.currentThread().setContextClassLoader(this.instanceClassLoader);
        }
    }
+
+    private static <T> Schema<T> getSinkSchema(Record<?> record, Class<T> clazz) {


we could move these 3 methods to some other class that does similar things, like TopicSchema.
but I have no strong opinion

We're doing these methods bc we don't have a topic. So it feels weird to put these in a class named TopicSchema

eolivelli · 2022-08-24T08:01:56Z

...ions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkSchemaInfoProvider.java

+import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+class SinkSchemaInfoProvider implements SchemaInfoProvider {


what about adding a javadoc here ?

eolivelli · 2022-08-24T08:04:24Z

...ions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkSchemaInfoProvider.java

+class SinkSchemaInfoProvider implements SchemaInfoProvider {
+
+  AtomicLong latestVersion = new AtomicLong(0);
+  ConcurrentHashMap<SchemaVersion, SchemaInfo> schemaInfos = new ConcurrentHashMap<>();


it is not very clear to me (by reading the code) who is populating these maps.
maybe we could add explicit accessors ?

I cut the code a little. Is it clearer now ?

nicoloboschi

LGTM

eolivelli

LGTM

cbornet · 2022-08-26T17:25:05Z

Following feedback, I renamed preprocess function to transform function.
cc @shibd @eolivelli @dave2wave @mattisonchao @nlu90 @nicoloboschi @HQebupt @lhotari @michaeljmarshall

If we detect that a Function is coupled with the Sink, we encode the record value with the Function schema and decode with the Sink schema. The Sink schema is created from the first schema type record received. At each record, a new schema version is created if it is the first time we get the hash of the record schema.

…D encoding

…in Sinks (apache#16740)

momo-jun · 2022-09-23T07:47:41Z

@cbornet It will help users a lot if you can add the docs for this feature. Do you have any plans on that?

cbornet · 2022-11-23T17:04:54Z

Transform functions are not part of 2.11 so we need to wait for the release before doing the doc...

momo-jun · 2022-11-29T07:22:42Z

@cbornet the milestone tag is still 2.11.
@Technoboy- just a double-check, is this feature going to be included in 2.11?

Technoboy- · 2022-12-01T03:37:20Z

@cbornet the milestone tag is still 2.11. @Technoboy- just a double-check, is this feature going to be included in 2.11?

Yes, not included in 2.11.

momo-jun · 2023-01-18T10:26:14Z

Transform functions are not part of 2.11 so we need to wait for the release before doing the doc...

Hi @cbornet, Since 2.11 has been released, now it's good timing to start adding the docs for this feature to the next version of docs. Feel free to share your plans and anything I can help with. Thank you.

cbornet marked this pull request as draft July 22, 2022 10:07

cbornet force-pushed the function-sink branch from f8b3c4f to 0c8feaf Compare July 22, 2022 15:14

cbornet marked this pull request as ready for review July 26, 2022 18:32

cbornet force-pushed the function-sink branch from 21f844e to 5221248 Compare July 26, 2022 21:11

eolivelli approved these changes Jul 27, 2022

View reviewed changes

shibd reviewed Jul 27, 2022

View reviewed changes

...ons/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RecordFunction.java Show resolved Hide resolved

cbornet force-pushed the function-sink branch from cdeac50 to 71f2758 Compare August 22, 2022 17:06

eolivelli changed the title ~~[feature][connectors] Add support for a preprocess Function in Sinks~~ [PIP193] [feature][connectors] Add support for a preprocess Function in Sinks Aug 24, 2022

eolivelli reviewed Aug 24, 2022

View reviewed changes

nicoloboschi approved these changes Aug 24, 2022

View reviewed changes

cbornet closed this Aug 24, 2022

cbornet reopened this Aug 24, 2022

cbornet force-pushed the function-sink branch from 7de1426 to c60ae14 Compare August 25, 2022 08:57

cbornet requested a review from eolivelli August 25, 2022 13:36

eolivelli approved these changes Aug 26, 2022

View reviewed changes

eolivelli changed the title ~~[PIP193] [feature][connectors] Add support for a preprocess Function in Sinks~~ [PIP-193] [feature][connectors] Add support for a preprocess Function in Sinks Aug 26, 2022

github-actions bot added the doc-required Your PR changes impact docs and you will update later. label Aug 26, 2022

cbornet changed the title ~~[PIP-193] [feature][connectors] Add support for a preprocess Function in Sinks~~ [PIP-193] [feature][connectors] Add support for a transform Function in Sinks Aug 26, 2022

cbornet force-pushed the function-sink branch 5 times, most recently from da621e0 to 84cf65c Compare August 29, 2022 15:36

cbornet closed this Aug 29, 2022

cbornet reopened this Aug 29, 2022

cbornet force-pushed the function-sink branch from 84cf65c to 7b71595 Compare August 29, 2022 22:45

cbornet closed this Aug 30, 2022

cbornet reopened this Aug 30, 2022

cbornet added 9 commits August 31, 2022 10:55

Add integration test for Sink with preprocessing function

e404a9e

Check that Sink preprocess function output type is a Record

e0a3ecd

Use a Sink with a type different than the Function in integration tests

9af64c6

Fix encoding/decoding when the record value is KeyValue with SEPARATE…

93d74d7

…D encoding

Minor improvements

1df5425

Rename preprocess function to transform function

483044d

Rename extra function to transform function

da3ed44

Fix: make --transform_function_id param non required

d260282

cbornet force-pushed the function-sink branch from 657c6b8 to d260282 Compare August 31, 2022 08:55

eolivelli merged commit dab0d1f into apache:master Aug 31, 2022

cbornet deleted the function-sink branch August 31, 2022 12:35

cbornet added a commit to cbornet/pulsar that referenced this pull request Aug 31, 2022

[PIP-193] [feature][connectors] Add support for a transform Function …

94281c1

…in Sinks (apache#16740)

nicoloboschi pushed a commit to datastax/pulsar that referenced this pull request Sep 1, 2022

[PIP-193] [feature][connectors] Add support for a transform Function …

b980f73

…in Sinks (apache#16740)

Technoboy- assigned cbornet Sep 23, 2022

Technoboy- added this to the 2.12.0 milestone Sep 23, 2022

Technoboy- added the area/connector label Sep 23, 2022

Technoboy- modified the milestones: 2.12.0, 2.11.0 Nov 14, 2022

Technoboy- modified the milestones: 2.11.0, 2.12.0 Dec 1, 2022

momo-jun mentioned this pull request May 8, 2023

Add doc for transform functions before sinks apache/pulsar-site#543

Merged

4 tasks

Anonymitaet added doc-complete Your PR changes impact docs and the related docs have been already added. and removed doc-required Your PR changes impact docs and you will update later. labels May 8, 2023

freeznet mentioned this pull request Dec 20, 2023

[feat] support transform function for sink connector streamnative/function-mesh#710

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[PIP-193] [feature][connectors] Add support for a transform Function in Sinks #16740

[PIP-193] [feature][connectors] Add support for a transform Function in Sinks #16740

cbornet commented Jul 22, 2022 •

edited

Loading

eolivelli left a comment

shibd Jul 27, 2022

eolivelli Jul 27, 2022

shibd Jul 27, 2022

cbornet Jul 27, 2022

shibd Jul 28, 2022

cbornet Jul 28, 2022

eolivelli Aug 24, 2022

cbornet Aug 25, 2022

eolivelli Aug 24, 2022

cbornet Aug 25, 2022

eolivelli Aug 24, 2022

cbornet Aug 25, 2022

eolivelli Aug 24, 2022

cbornet Aug 25, 2022

eolivelli Aug 26, 2022

eolivelli Aug 24, 2022

cbornet Aug 25, 2022

eolivelli Aug 26, 2022

nicoloboschi left a comment

eolivelli left a comment

cbornet commented Aug 26, 2022

momo-jun commented Sep 23, 2022

cbornet commented Nov 23, 2022

momo-jun commented Nov 29, 2022

Technoboy- commented Dec 1, 2022

momo-jun commented Jan 18, 2023

	if (result.getUserException() != null) {
	Exception t = result.getUserException();
	log.warn("Encountered exception when processing message {}",
	srcRecord, t);
	stats.incrUserExceptions(t);
	srcRecord.fail();
	} else {

[PIP-193] [feature][connectors] Add support for a transform Function in Sinks #16740

[PIP-193] [feature][connectors] Add support for a transform Function in Sinks #16740

Conversation

cbornet commented Jul 22, 2022 • edited Loading

Motivation

Modifications

Verifying this change

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

Documentation

eolivelli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nicoloboschi left a comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

cbornet commented Aug 26, 2022

momo-jun commented Sep 23, 2022

cbornet commented Nov 23, 2022

momo-jun commented Nov 29, 2022

Technoboy- commented Dec 1, 2022

momo-jun commented Jan 18, 2023

cbornet commented Jul 22, 2022 •

edited

Loading