[PIP 86][Function] Preload and release of external resources #11112

wuchenxi123 · 2021-06-26T13:38:40Z

(If this PR fixes a github issue, please add Fixes #<xyz>.)

Fixes #

(or if this PR is one task of a github issue, please add Master Issue: #<xyz> to link to the master issue.)

Master Issue: #

Motivation

Before Function is started, some resources only need to be initialized once, and there is no need to make various judgments in the process() method of the Function interface
After the Function is closed, in the process of using process(), the referenced external resources need to be manually closed(), and they need to be released separately in the close() of javaInstance

Modifications

Before starting, some resources only need to be loaded once
When Function Instance manually releases some resources, users need to implement the release method and provide support

@slf4j
public class HelloWorldFunction implements Function<String, Void> {
@OverRide
public String process(byte[] input, Context context) {
}
}
eg:
@slf4j
public class HelloWorldFunction implements Function<String, Void>,Hook {
@OverRide
public String process(byte[] input, Context context) {

}
/**
	resource init
*/
@Override
public void preProcess(Context context) throws Exception {
	database Connection
	other resource initialize
}
/**
	resource close
*/
@Override
public void postProcess() {
	connection close
	call the callback function
}

}

Provide an interface for initialization and unified release of external resources for users to use, easy to manage and expand
The order of the release of external resources is controlled by the pulsar function. Users only need to pay attention to the event processing logic. The code is clearer and more concise, and the hierarchy is clear

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:
yes

(example:)

The initialization and release functions have been applied and verified in the online project

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): (yes / no)
The public API: (yes / no) yes
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

Does this pull request introduce a new feature? (yes / no) yes
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
If a feature is not applicable for documentation, explain why?
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

BewareMyPower

Could you explain why did you swallow the exception in close() while fail fast in setup()? For your case, preProcess should not throw a checked exception.

In addition, IMO it's not a good practice to just use Exception as the checked exception but not the subclass of Exception.

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

wuchenxi123 · 2021-06-27T13:27:51Z

Could you explain why did you swallow the exception in close() while fail fast in setup()? For your case, preProcess should not throw a checked exception.

In addition, IMO it's not a good practice to just use Exception as the checked exception but not the subclass of Exception.

Accept your suggestion, I have corrected your problem.Please codereview

freeznet

I like the idea of this PR adding two steps (init and close) into the pulsar function, I have left some comments as the review, PTAL.

freeznet · 2021-06-27T13:46:19Z

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

@@ -160,7 +179,22 @@ private void processAsyncResults(BiConsumer<Record, JavaExecutionResult> resultC
    @Override
    public void close() {
        context.close();
-        executor.shutdown();


why remove this line?

I have corrected it

freeznet · 2021-06-27T14:40:03Z

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

+     *
+     * @throws Exception
+     */
+    void preProcess(Context context) throws RuntimeException;


preProcess and postProcess is likely to be invoked with each message each time before and after process, so the naming is kind confuse from my understanding.

I have corrected it

freeznet · 2021-06-27T14:46:21Z

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

+                throw new RuntimeException("function preProcess occurred exception", e);
+            }
+        }
+        if (null != javaUtilFunction && javaUtilFunction instanceof Hook) {


according to the docs, java.util.function.Function is not intend to interact with Context, so we should avoid to passing Context to java.util.function.Function.

I have corrected it

BewareMyPower · 2021-06-28T01:03:36Z

You may misunderstand my comment.

IMO it's not a good practice to just use Exception as the checked exception but not the subclass of Exception.

I mean use a derived class of Exception as the checked exception. You replaced Exception with RuntimeException, which is worse because RuntimeException is an unchecked exception. Take org.apache.pulsar.metadata.api.MetadataSerde as an example:

public interface MetadataSerde<T> {

    byte[] serialize(T value) throws IOException;

    T deserialize(byte[] content) throws IOException;
}

It checks IOException that derives from Exception. You can also some many other interfaces in Pulsar that checks PulsarAdminException or PulsarClientException, not Exception directly.

Though IMO it's a better practice, using Exception directly is also okay but don't use RuntimeException as the checked exception.

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

wuchenxi123 · 2021-06-28T15:14:09Z

You may misunderstand my comment.

IMO it's not a good practice to just use Exception as the checked exception but not the subclass of Exception.

I mean use a derived class of Exception as the checked exception. You replaced Exception with RuntimeException, which is worse because RuntimeException is an unchecked exception. Take org.apache.pulsar.metadata.api.MetadataSerde as an example:
public interface MetadataSerde<T> {

    byte[] serialize(T value) throws IOException;

    T deserialize(byte[] content) throws IOException;
}
It checks IOException that derives from Exception. You can also some many other interfaces in Pulsar that checks PulsarAdminException or PulsarClientException, not Exception directly.

Though IMO it's a better practice, using Exception directly is also okay but don't use RuntimeException as the checked exception.

I have modified it, please check

nlu90 · 2021-07-02T21:18:08Z

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

+ * Initial and close of external resources
+ */
+@InterfaceAudience.Public
+public interface Hook {


I suggest this new api extends the Function interface since it's strongly coupled with the Function api and should not be used individually in other cases.

public interface HookFunction extends Function { void setup(); void cleanup(); }

And we should keep java.util.Function as it is to keep its simplicity.

I have modified it, please check

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

# Conflicts: # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

eolivelli

I find this feature very useful.

Can you please start aa discussion on dev@pulsar.apache.org in order to show your needs and how you want to address them?

The mailing list is the place in which the community discusses about new features

wuchenxi123 · 2021-07-06T12:35:14Z

I find this feature very useful.

Can you please start aa discussion on dev@pulsar.apache.org in order to show your needs and how you want to address them?

The mailing list is the place in which the community discusses about new features

I have sent an email, please check

eolivelli · 2021-07-06T12:37:18Z

I have sent an email, please check

thanks

nlu90 · 2021-07-06T18:36:25Z

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

+        if (null != javaUtilFunction && javaUtilFunction instanceof HookFunction) {
+            try {
+                ((HookFunction) javaUtilFunction).setup();
+            } catch (Exception e) {
+                log.error("setup error:", e);
+                throw e;
+            }
+        }


I suggest we keep java.util.Function as simple as possible, and not add hook method for it

nlu90 · 2021-07-06T18:37:22Z

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

+        if (null != javaUtilFunction && javaUtilFunction instanceof HookFunction) {
+            try {
+                ((HookFunction) javaUtilFunction).cleanup();
+            } catch (Exception e) {


I suggest we keep java.util.Function as simple as possible, and not add hook method for it

I think HookFunction can provide users with additional choices, and it also makes the responsibilities of each method of Function more single and clear, please think again

nlu90 · 2021-07-06T18:39:11Z

Sorry, I accidentally add the comment with approve.

Please let me know if you update the PR and I'll review again.

jerrypeng

Some of my thoughts:

If we are going to introduce new methods or stages in the lifecycle of a function such as "open" and "close", I would recommend we follow what we already have for Pulsar IO sources and Sink so that we are consistent.
I am also not a particular fan of the name "HookFunction". Can we call it "RichFunction"? This naming is consistent with other projects such as Apache Storm and Flink.
In regards to the topic

If this rule must be followed, then one possible way is we define a separate HookFunction interface which includes setup, process, tearDown methods, and add support in the new runtime to handle this interface. Just as currently the runtime handles pulsar.Function and java.util.Function separately. The old runtime will not be able to recognize the new interface and thus won't execute functions which need initialization.

I would rather not define a completely new interface. Ideally we have a "RichFunction" interface that extends the existing. So that we can minimize interface duplication and if we want to introduce new interfaces in the future we can build on top of that. In regards to the issue about the old runtime executing a function that implements the new interface, is it sufficient to document somewhere that the worker has to be updated to a certain version to run the functions that implement the new interface?

wuchenxi123 · 2021-07-10T05:45:50Z

Some of my thoughts:

If we are going to introduce new methods or stages in the lifecycle of a function such as "open" and "close", I would recommend we follow what we already have for Pulsar IO sources and Sink so that we are consistent.

I am also not a particular fan of the name "HookFunction". Can we call it "RichFunction"? This naming is consistent with other projects such as Apache Storm and Flink.

In regards to the topic

If this rule must be followed, then one possible way is we define a separate HookFunction interface which includes setup, process, tearDown methods, and add support in the new runtime to handle this interface. Just as currently the runtime handles pulsar.Function and java.util.Function separately. The old runtime will not be able to recognize the new interface and thus won't execute functions which need initialization.

I would rather not define a completely new interface. Ideally we have a "RichFunction" interface that extends the existing. So that we can minimize interface duplication and if we want to introduce new interfaces in the future we can build on top of that. In regards to the issue about the old runtime executing a function that implements the new interface, is it sufficient to document somewhere that the worker has to be updated to a certain version to run the functions that implement the new interface?

I agree with your point of view, and made the corresponding changes, please codereview

wangjialing218 · 2021-10-09T03:10:51Z

@BewareMyPower @nlu90 @jerrypeng @eolivelli PTAL when you have time since Pulsar 2.9.0 release cut is coming.

nlu90 · 2021-11-29T19:16:43Z

@jerrypeng @eolivelli Should we resume the discussion of this useful feature?

eolivelli

I believe that there is value in this work.

Is the discussion stale?

What's the status of the PIP?

Are we adoption the new process? That is to have a github issue with the design doc? That would help people to review the design decisions.

Feel free to ping me on slack if you need some guidance

eolivelli · 2021-11-29T20:28:32Z

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

        context.close();
        executor.shutdown();
+        if (null != function && function instanceof RichFunction) {


Actually you don't need the null check as null is never a instance of a class

eolivelli · 2021-11-29T20:29:44Z

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

+     *
+     * @throws Exception
+     */
+    void setup() throws Exception;


I would pass some Context in order to let the Function initialise using the information about the env.

nlu90 · 2021-11-29T23:51:54Z

@wuchenxi123 Do you want to provide some update to Enrico's above questions?

@wuchenxi123

…ernal resources (#13205) ### Motivation This patch is based on #11112, I have talked with @wuchenxi123 and I will continue the work for this feature. ### Modifications Introduce `RichFunction` interface extends `Function`, which provide `setup` and `tearDown` API. `setup` is called only once when function instance started. `tearDown` is called when function instance closed. User could use these interface to initialize and release their external resources such as RedisClient. ### Does this pull request potentially affect one of the following parts: *If `yes` was chosen, please highlight the changes* - The public API: ( **yes**)

@wuchenxi123

…ernal resources (apache#13205) ### Motivation This patch is based on apache#11112, I have talked with @wuchenxi123 and I will continue the work for this feature. ### Modifications Introduce `RichFunction` interface extends `Function`, which provide `setup` and `tearDown` API. `setup` is called only once when function instance started. `tearDown` is called when function instance closed. User could use these interface to initialize and release their external resources such as RedisClient. ### Does this pull request potentially affect one of the following parts: *If `yes` was chosen, please highlight the changes* - The public API: ( **yes**)

pulsar-function:Preload and release of external resources

f51c76a

BewareMyPower reviewed Jun 27, 2021

View reviewed changes

BewareMyPower assigned wuchenxi123 Jun 27, 2021

BewareMyPower added the area/function label Jun 27, 2021

chenlei added 2 commits June 27, 2021 21:21

pulsar-function:Preload and release of external resources

ff26202

Merge remote-tracking branch 'chenlei/develop' into develop

c5e6055

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

freeznet reviewed Jun 27, 2021

View reviewed changes

chenlei added 4 commits June 28, 2021 23:00

Merge remote-tracking branch 'chenlei/develop' into develop

b631bfb

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

Merge remote-tracking branch 'chenlei/develop' into develop

93c4300

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

Merge remote-tracking branch 'chenlei/develop' into develop

e30457f

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

Merge remote-tracking branch 'chenlei/develop' into develop

b655851

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

nlu90 requested changes Jul 2, 2021

View reviewed changes

chenlei added 4 commits July 4, 2021 19:29

Merge remote-tracking branch 'chenlei/develop' into develop

f355667

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

Merge remote-tracking branch 'chenlei/develop' into develop

c6b2d92

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

Merge remote-tracking branch 'chenlei/develop' into develop

c1b2c51

# Conflicts: # pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Hook.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

Merge remote-tracking branch 'chenlei/develop' into develop

c7556d4

# Conflicts: # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java

eolivelli requested changes Jul 4, 2021

View reviewed changes

nlu90 approved these changes Jul 6, 2021

View reviewed changes

jerrypeng requested changes Jul 9, 2021

View reviewed changes

Modifications to jerrypeng's comments

9d003df

wuchenxi123 changed the title ~~pulsar-function:Preload and release of external resources~~ 【PIP 86】【Function】Preload and release of external resources Oct 9, 2021

wuchenxi123 changed the title ~~【PIP 86】【Function】Preload and release of external resources~~ [PIP 86][Function] Preload and release of external resources Oct 9, 2021

eolivelli requested changes Nov 29, 2021

View reviewed changes

wangjialing218 mentioned this pull request Dec 9, 2021

[PIP 86][Function] add funtion api to support preload and release external resources #13205

Merged

1 task

wuchenxi123 closed this Jan 21, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[PIP 86][Function] Preload and release of external resources #11112

[PIP 86][Function] Preload and release of external resources #11112

wuchenxi123 commented Jun 26, 2021

BewareMyPower left a comment

wuchenxi123 commented Jun 27, 2021 •

edited

Loading

freeznet left a comment

freeznet Jun 27, 2021

wuchenxi123 Jun 28, 2021

freeznet Jun 27, 2021

wuchenxi123 Jun 28, 2021

freeznet Jun 27, 2021

wuchenxi123 Jun 28, 2021

BewareMyPower commented Jun 28, 2021

wuchenxi123 commented Jun 28, 2021

nlu90 Jul 2, 2021

wuchenxi123 Jul 4, 2021

eolivelli left a comment

wuchenxi123 commented Jul 6, 2021

eolivelli commented Jul 6, 2021

nlu90 Jul 6, 2021

nlu90 Jul 6, 2021

wuchenxi123 Jul 7, 2021

nlu90 commented Jul 6, 2021

jerrypeng left a comment

wuchenxi123 commented Jul 10, 2021

wangjialing218 commented Oct 9, 2021

nlu90 commented Nov 29, 2021

eolivelli left a comment

eolivelli Nov 29, 2021

eolivelli Nov 29, 2021

nlu90 commented Nov 29, 2021

[PIP 86][Function] Preload and release of external resources #11112

[PIP 86][Function] Preload and release of external resources #11112

Conversation

wuchenxi123 commented Jun 26, 2021

Motivation

Modifications

Verifying this change

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

Documentation

BewareMyPower left a comment

Choose a reason for hiding this comment

wuchenxi123 commented Jun 27, 2021 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BewareMyPower commented Jun 28, 2021

wuchenxi123 commented Jun 28, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

wuchenxi123 commented Jul 6, 2021

eolivelli commented Jul 6, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nlu90 commented Jul 6, 2021

jerrypeng left a comment

Choose a reason for hiding this comment

wuchenxi123 commented Jul 10, 2021

wangjialing218 commented Oct 9, 2021

nlu90 commented Nov 29, 2021

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

nlu90 commented Nov 29, 2021

wuchenxi123 commented Jun 27, 2021 •

edited

Loading