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

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

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.api;
+
+import org.apache.pulsar.common.classification.InterfaceAudience;
+
+/**
+ * An interface for hook
+ * Preload and release of external resources
+ */
+@InterfaceAudience.Public
+public interface Hook {
+
+    /**
+     * Pre-Process Function Hook
+     *
+     * @throws Exception
+     */
+    void preProcess(Context context) throws Exception;
+
+
+    /**
+     * Post-Process Function Hook
+     *
+     * @throws Exception
+     */
+    void postProcess() throws Exception;
+}

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

@@ -32,6 +32,7 @@
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.functions.api.Function;
+import org.apache.pulsar.functions.api.Hook;
 import org.apache.pulsar.functions.api.Record;
 
 /**
@@ -74,6 +75,24 @@ public JavaInstance(ContextImpl contextImpl, Object userClassObject, InstanceCon
         }
     }
 
+    public void setup() {
+        if (null != function && function instanceof Hook) {
+            try {
+                ((Hook) function).preProcess(context);
+            } catch (Exception e) {
+                log.error("setup error:", e);
+                throw new RuntimeException("function preProcess occurred exception", e);
+            }
+        }
+        if (null != javaUtilFunction && javaUtilFunction instanceof Hook) {
+            try {
+                ((Hook) javaUtilFunction).preProcess(context);
+            } catch (Exception e) {
+                log.error("setup error:", e);
+                throw new RuntimeException("javaUtilFunction preProcess occurred exception", e);
+            }
+        }
+    }
     @VisibleForTesting
     public JavaExecutionResult handleMessage(Record<?> record, Object input) {
         return handleMessage(record, input, (rec, result) -> {}, cause -> {});
@@ -160,6 +179,20 @@ private void processAsyncResults(BiConsumer<Record, JavaExecutionResult> resultC
     @Override
     public void close() {
         context.close();
+        if (null != function && function instanceof Hook) {
+            try {
+                ((Hook) function).postProcess();
+            } catch (Exception e) {
+                log.error("function postProcess occurred exception", e);
+            }
+        }
+        if (null != javaUtilFunction && javaUtilFunction instanceof Hook) {
+            try {
+                ((Hook) javaUtilFunction).postProcess();
+            } catch (Exception e) {
+                log.error("javaUtilFunction postProcess occurred exception", e);
+            }
+        }
         executor.shutdown();
     }
 

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

@@ -221,7 +221,7 @@ synchronized private void setup() throws Exception {
         setupLogHandler();
 
         javaInstance = new JavaInstance(contextImpl, object, instanceConfig);
-
+        javaInstance.setup();
         // to signal member variables are initialized
         isInitialized = true;
     }