[improve][function] Support Record<?> as Function output type #16041

cbornet · 2022-06-13T13:52:35Z

Motivation

Currently, when a user wants to dynamically set the output topic, the message properties or change the output schema in a Function, the only possibility is to create a Function that returns Void , use the Context and manually create a message with Context::newOutputMessage. The TypedMessageBuilderPublish Function in pulsar-functions-api-examples shows how to do that.
This way of doing is not intuitive and it would be better to return a structure like Record that carries this info.

Modifications

This PR adds support for returning Record in a Function.

In JavaInstanceRunnable::sendOutputMessage, we check the type of the output object and if it's of type Record we create a TargetSinkRecord instead of a SinkRecord. TargetSinkRecord uses the info from the output record in the various Record methods that are used by the Sink when creating the output message.
When registering the Function, in getFunctionTypes, if the output type is a Record, we get the wrapped type as type for the Sink.
A utility method newOutputRecordBuilder is added to Context that returns a FunctionRecord builder initialized with the info from the source record. The builder methods can then be used to override these values as needed.
A RecordFunction is added in the Function examples to demonstrate the use of this feature
An integration test is added PulsarFunctionsJavaTest

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

Run PulsarFunctionsJavaTest:: testRecordFunctionTest

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

yes

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API: (yes / no)
Yes.
- adds Record<> as a possible return type for Functions
- adds a newOutputRecordBuilder method to Context
The schema: (yes / no / don't know)
The default values of configurations: (yes / no)
The wire protocol: (yes / no)
The rest endpoints: (yes / no)
The admin cli options: (yes / no)
Anything that affects deployment: (yes / no / don't know)

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)

lhotari

LGTM. Is this change ABI compatible so that existing user created Pulsar Functions implementations can be run without recompiling when upgrading to a version that includes this change?

eolivelli

Overall LGTM

I left some feedback

eolivelli · 2022-06-14T07:42:35Z

...r-functions/api-java/src/main/java/org/apache/pulsar/functions/api/utils/FunctionRecord.java

+        currentRecord.getPartitionIndex().ifPresent(builder::partitionIndex);
+        currentRecord.getRecordSequence().ifPresent(builder::recordSequence);
+
+        // TODO: add message


if you want to do this, please create a ticket and link it, otherwise please remove this "TODO"
TODOs are usually some kind of code smell, especially in a big open source project like Pulsar.

we could explain why the "message" is not available here

Sorry, I forgot to remove that one.
I totally agree that TODOs must not hit main branch 😄

eolivelli · 2022-06-14T07:44:17Z

...rc/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaTest.java

@@ -166,7 +166,7 @@ public void testSlidingCountWindowTest() throws Exception {
    @Test(groups = {"java_function", "function"})
    public void testMergeFunctionTest() throws Exception {
 	    testMergeFunction();
-   }
+    }


nit: remove space

This is to align with the rest of the function on 4 spaces.
But this whole class is misaligned...

I realigned completely this block but I guess the full class should be realigned.
Maybe better to do it in another PR ?

yes, this PR should focus on the feature and not on code clean up

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java

eolivelli · 2022-06-14T07:48:43Z

...r-functions/api-java/src/main/java/org/apache/pulsar/functions/api/utils/FunctionRecord.java

+    private final Integer partitionIndex;
+    private final Long recordSequence;
+
+    public static <T> FunctionRecord.FunctionRecordBuilder<T> from(Context context) {


we need unit test coverage for this method (ensure that every field is properly handled)

eolivelli · 2022-06-14T07:49:06Z

pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java

@@ -161,4 +162,11 @@ public interface Context extends BaseContext {
     * @throws PulsarClientException
     */
    <X> ConsumerBuilder<X> newConsumerBuilder(Schema<X> schema) throws PulsarClientException;
+
+    /**
+     * Create a FunctionRecordBuilder.


please explain a little bit how this is supposed to be used.

cbornet · 2022-06-14T13:35:01Z

Is this change ABI compatible so that existing user created Pulsar Functions implementations can be run without recompiling when upgrading to a version that includes this change?

@lhotari I think it is since we only add a method to the Context API.
SinkRecord is sometimes used outside of the package (and probably shouldn't) in tests but I think extracting some methods to a super class is OK for binary compat. Or is it not ?
Do you see other things that could break ?

* Rename TargetSinkRecord to OutputRecordSinkRecord * Add UT for FunctionCommon::getFunctionTypes * Add UT for Context::newOutputRecordBuilder * Add some javadoc * Always set message properties for OutputRecordSinkRecord

eolivelli

LGTM

codelipenghui · 2022-06-15T14:39:02Z

@nlu90 Please help review this PR.

cbornet · 2022-06-15T15:09:56Z

/pulsarbot rerun-failure-checks

freeznet

LGTM, but I do have a question, it seems this method can be a replacement of newOutputMessage, so is it possible to deprecate newOutputMessage? if not, could you please add some more context comparing newOutputRecordBuilder and newOutputMessage? thanks.

nlu90 · 2022-06-16T04:24:23Z

...r-functions/api-java/src/main/java/org/apache/pulsar/functions/api/utils/FunctionRecord.java

+     * @param <T> type of Record to build
+     * @return a Record builder initialised with values from the Function Context
+     */
+     public static <T> FunctionRecord.FunctionRecordBuilder<T> from(Context context) {


Could you help me understand why the whole context is passed here, instead of the currentRecord?

We set the destinationTopic from the Context getOutputTopic

nlu90 · 2022-06-16T04:25:52Z

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

@@ -481,6 +482,11 @@ public <T> ConsumerBuilder<T> newConsumerBuilder(Schema<T> schema) throws Pulsar
        return this.client.newConsumer(schema);
    }

+    @Override
+    public <X> FunctionRecord.FunctionRecordBuilder<X> newOutputRecordBuilder() {


Instead of adding a new API here, do you think about creating a util class with a record generation method?

I think it is better to not add a general purpose API to build "Record" instances.
A Record is like a Message and you cannot build a Message using the client API or a utility class.
The record may be tied to some internal context or pre-configured (as it does currently in this PR).

So it is better to use Context as a starting point for building a new record, the same way it happens with newOutputMessage()

cbornet · 2022-06-16T08:23:04Z

LGTM, but I do have a question, it seems this method can be a replacement of newOutputMessage, so is it possible to deprecate newOutputMessage? if not, could you please add some more context comparing newOutputRecordBuilder and newOutputMessage? thanks.

That's 2 different approaches. newOutputRecordBuilder is useful if you design a Function that returns Record. newOutputMessage is more generic and could be used to send multiple messages for instance. It also gives more control on the message as you can control things such as deliverAt/deliverAfter.

…#16041)

momo-jun · 2022-06-21T01:42:34Z

Hi @cbornet , will you update relevant docs in a follow-up PR?

tisonkun · 2022-07-21T10:30:18Z

@cbornet @momo-jun is there any existing doc we can improve?

Perhaps we should place the RecordFunction demo under "Pulsar Functions :: How to: Develop" entry with a new page?

tisonkun · 2023-01-12T06:55:34Z

...r-functions/api-java/src/main/java/org/apache/pulsar/functions/api/utils/FunctionRecord.java

+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Record;
+
+@Builder(builderMethodName = "")


Is it necessary to suppress generating the builder method?

[improve][function] Support Record<?> as Function output type

eeaa729

cbornet force-pushed the function-record branch from 047ce38 to eeaa729 Compare June 13, 2022 15:04

dave2wave requested review from eolivelli and lhotari June 13, 2022 19:41

Anonymitaet added the doc-required Your PR changes impact docs and you will update later. label Jun 14, 2022

lhotari approved these changes Jun 14, 2022

View reviewed changes

eolivelli requested changes Jun 14, 2022

View reviewed changes

cbornet requested a review from eolivelli June 14, 2022 14:34

Changes following review

a56d6c1

* Rename TargetSinkRecord to OutputRecordSinkRecord * Add UT for FunctionCommon::getFunctionTypes * Add UT for Context::newOutputRecordBuilder * Add some javadoc * Always set message properties for OutputRecordSinkRecord

cbornet force-pushed the function-record branch from b396322 to a56d6c1 Compare June 14, 2022 14:39

eolivelli approved these changes Jun 15, 2022

View reviewed changes

codelipenghui requested review from freeznet and merlimat June 15, 2022 14:38

codelipenghui assigned cbornet Jun 15, 2022

codelipenghui added this to the 2.11.0 milestone Jun 15, 2022

codelipenghui added area/function type/feature The PR added a new feature or issue requested a new feature labels Jun 15, 2022

freeznet approved these changes Jun 16, 2022

View reviewed changes

nlu90 approved these changes Jun 16, 2022

View reviewed changes

eolivelli merged commit b3c5191 into apache:master Jun 17, 2022

cbornet deleted the function-record branch June 17, 2022 11:37

cbornet added a commit to cbornet/pulsar that referenced this pull request Jun 18, 2022

[improve][function] Support Record<?> as Function output type (apache…

0d79b74

…#16041)

cbornet added a commit to cbornet/pulsar that referenced this pull request Jun 18, 2022

[improve][function] Support Record<?> as Function output type (apache…

d8a3a0e

…#16041)

cbornet mentioned this pull request Jun 18, 2022

[improve][function] Support Record<?> as WindowFunction output type #16129

Merged

5 tasks

cbornet added a commit to datastax/pulsar that referenced this pull request Jun 20, 2022

[improve][function] Support Record<?> as Function output type (apache…

7e7a95b

…#16041)

momo-jun mentioned this pull request Jul 27, 2022

[improve][doc] Add doc on Record as Function output type #16816

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 Aug 2, 2022

tisonkun reviewed Jan 12, 2023

View reviewed changes

tisonkun mentioned this pull request Jan 12, 2023

[fix][doc] Remove lombok plugin and define FunctionRecordBuilder explicitly #19202

Merged

14 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve][function] Support Record<?> as Function output type #16041

[improve][function] Support Record<?> as Function output type #16041

cbornet commented Jun 13, 2022 •

edited

Loading

lhotari left a comment

eolivelli left a comment

eolivelli Jun 14, 2022

cbornet Jun 14, 2022

eolivelli Jun 14, 2022

cbornet Jun 14, 2022

cbornet Jun 14, 2022

eolivelli Jun 16, 2022

eolivelli Jun 14, 2022

cbornet Jun 14, 2022

eolivelli Jun 14, 2022

cbornet Jun 14, 2022

cbornet commented Jun 14, 2022

eolivelli left a comment

codelipenghui commented Jun 15, 2022

cbornet commented Jun 15, 2022

freeznet left a comment

nlu90 Jun 16, 2022

cbornet Jun 16, 2022

nlu90 Jun 16, 2022

eolivelli Jun 16, 2022

cbornet commented Jun 16, 2022 •

edited

Loading

momo-jun commented Jun 21, 2022

tisonkun commented Jul 21, 2022

tisonkun Jan 12, 2023

[improve][function] Support Record<?> as Function output type #16041

[improve][function] Support Record<?> as Function output type #16041

Conversation

cbornet commented Jun 13, 2022 • edited Loading

Motivation

Modifications

Verifying this change

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

Documentation

lhotari left a comment

Choose a reason for hiding this comment

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

cbornet commented Jun 14, 2022

eolivelli left a comment

Choose a reason for hiding this comment

codelipenghui commented Jun 15, 2022

cbornet commented Jun 15, 2022

freeznet 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

cbornet commented Jun 16, 2022 • edited Loading

momo-jun commented Jun 21, 2022

tisonkun commented Jul 21, 2022

Choose a reason for hiding this comment

cbornet commented Jun 13, 2022 •

edited

Loading

cbornet commented Jun 16, 2022 •

edited

Loading