Support different docker images in Kubernetes runtime of Pulsar Funct…

…ions (apache#6752)

## Motivation
Provide support to configure different docker images for go, python, java function

## Modifications
Add javaFunctionDockerImageName, pythonFunctionDockerImageName, and goFunctionDockerImageName in KubernetesRuntimeFactoryConfig.
Add unit test case
Update docs of functions-runtime.md

conf/functions_worker.yml

@@ -141,6 +141,14 @@ functionRuntimeFactoryConfigs:
 #  jobNamespace:
 #  # the docker image to run function instance. by default it is `apachepulsar/pulsar`
 #  pulsarDockerImageName:
+#  # the docker image to run function instance according to different configurations provided by users.
+#  # By default it is `apachepulsar/pulsar`.
+#  # e.g:
+#  # functionDockerImages:
+#  #   JAVA: JAVA_IMAGE_NAME
+#  #   PYTHON: PYTHON_IMAGE_NAME
+#  #   GO: GO_IMAGE_NAME
+#  functionDockerImages:
 #  # the root directory of pulsar home directory in `pulsarDockerImageName`. by default it is `/pulsar`.
 #  # if you are using your own built image in `pulsarDockerImageName`, you need to set this setting accordingly
 #  pulsarRootDir:

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

@@ -48,4 +48,8 @@ public class InstanceConfig {
     public String getInstanceName() {
         return "" + instanceId;
     }
+
+    public FunctionDetails getFunctionDetails() {
+        return functionDetails;
+    }
 }

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntime.java

@@ -85,6 +85,7 @@
 
 import static java.net.HttpURLConnection.HTTP_CONFLICT;
 import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
+import static org.apache.commons.lang3.StringUtils.defaultIfBlank;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.pulsar.functions.auth.FunctionAuthUtils.getFunctionAuthData;
 import static org.apache.pulsar.functions.utils.FunctionCommon.roundDecimal;
@@ -130,6 +131,7 @@ public class KubernetesRuntime implements Runtime {
     private InstanceConfig instanceConfig;
     private final String jobNamespace;
     private final Map<String, String> customLabels;
+    private final Map<String, String> functionDockerImages;
     private final String pulsarDockerImageName;
     private final String imagePullPolicy;
     private final String pulsarRootDir;
@@ -156,6 +158,7 @@ public class KubernetesRuntime implements Runtime {
                       String pythonDependencyRepository,
                       String pythonExtraDependencyRepository,
                       String pulsarDockerImageName,
+                      Map<String, String> functionDockerImages,
                       String imagePullPolicy,
                       String pulsarRootDir,
                       InstanceConfig instanceConfig,
@@ -185,6 +188,7 @@ public class KubernetesRuntime implements Runtime {
         this.instanceConfig = instanceConfig;
         this.jobNamespace = jobNamespace;
         this.customLabels = customLabels;
+        this.functionDockerImages = functionDockerImages;
         this.pulsarDockerImageName = pulsarDockerImageName;
         this.imagePullPolicy = imagePullPolicy;
         this.pulsarRootDir = pulsarRootDir;
@@ -977,8 +981,35 @@ private List<V1Toleration> getTolerations() {
     V1Container getFunctionContainer(List<String> instanceCommand, Function.Resources resource) {
         final V1Container container = new V1Container().name(PULSARFUNCTIONS_CONTAINER_NAME);
 
-        // set up the container images
-        container.setImage(pulsarDockerImageName);
+        Function.FunctionDetails.Runtime runtime = instanceConfig.getFunctionDetails().getRuntime();
+
+        String imageName = null;
+        if (functionDockerImages != null) {
+            switch (runtime) {
+                case JAVA:
+                    if (functionDockerImages.get("JAVA") != null) {
+                        imageName = functionDockerImages.get("JAVA");
+                        break;
+                    }
+                case PYTHON:
+                    if (functionDockerImages.get("PYTHON") != null) {
+                        imageName = functionDockerImages.get("PYTHON");
+                        break;
+                    }
+                case GO:
+                    if (functionDockerImages.get("GO") != null) {
+                        imageName = functionDockerImages.get("GO");
+                        break;
+                    }
+                default:
+                    imageName = pulsarDockerImageName;
+                    break;
+            }
+            container.setImage(imageName);
+        } else {
+            container.setImage(pulsarDockerImageName);
+        }
+
         container.setImagePullPolicy(imagePullPolicy);
 
         // set up the container command

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactory.java

@@ -67,6 +67,7 @@ public class KubernetesRuntimeFactory implements RuntimeFactory {
     private String k8Uri;
     private String jobNamespace;
     private String pulsarDockerImageName;
+    private Map<String, String> functionDockerImages;
     private String imagePullPolicy;
     private String pulsarRootDir;
     private String configAdminCLI;
@@ -143,6 +144,7 @@ public void initialize(WorkerConfig workerConfig, AuthenticationConfig authentic
         } else {
             this.pulsarDockerImageName = "apachepulsar/pulsar";
         }
+        this.functionDockerImages = factoryConfig.getFunctionDockerImages();
         if (!isEmpty(factoryConfig.getImagePullPolicy())) {
             this.imagePullPolicy = factoryConfig.getImagePullPolicy();
         } else {
@@ -272,6 +274,7 @@ public KubernetesRuntime createContainer(InstanceConfig instanceConfig, String c
             pythonDependencyRepository,
             pythonExtraDependencyRepository,
             pulsarDockerImageName,
+            functionDockerImages,
             imagePullPolicy,
             pulsarRootDir,
             instanceConfig,

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactoryConfig.java

@@ -43,6 +43,12 @@ public class KubernetesRuntimeFactoryConfig {
     )
     protected String pulsarDockerImageName;
 
+    @FieldContext(
+            doc = "The function docker images used to run function instance according to different "
+                    + "configurations provided by users. By default it is `apachepulsar/pulsar`"
+    )
+    protected Map<String, String> functionDockerImages;
+
     @FieldContext(
             doc = "The image pull policy for image used to run function instance. By default it is `IfNotPresent`"
     )

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactoryTest.java

@@ -153,6 +153,7 @@ KubernetesRuntimeFactory createKubernetesRuntimeFactory(String extraDepsDir,
         kubernetesRuntimeFactoryConfig.setK8Uri(null);
         kubernetesRuntimeFactoryConfig.setJobNamespace(null);
         kubernetesRuntimeFactoryConfig.setPulsarDockerImageName(null);
+        kubernetesRuntimeFactoryConfig.setFunctionDockerImages(null);
         kubernetesRuntimeFactoryConfig.setImagePullPolicy(null);
         kubernetesRuntimeFactoryConfig.setPulsarRootDir(pulsarRootDir);
         kubernetesRuntimeFactoryConfig.setSubmittingInsidePod(false);
@@ -367,10 +368,15 @@ public void testDynamicConfigMapLoading() throws Exception {
     private KubernetesRuntimeFactory getKuberentesRuntimeFactory() {
         KubernetesRuntimeFactory kubernetesRuntimeFactory = new KubernetesRuntimeFactory();
         WorkerConfig workerConfig = new WorkerConfig();
+        Map<String, String> imageNames = new HashMap<>();
+        imageNames.put("JAVA", "test-java-function-docker-image");
+        imageNames.put("PYTHON", "test-python-function-docker-image");
+        imageNames.put("GO", "test-go-function-docker-image");
         KubernetesRuntimeFactoryConfig kubernetesRuntimeFactoryConfig = new KubernetesRuntimeFactoryConfig();
         kubernetesRuntimeFactoryConfig.setK8Uri("test_k8uri");
         kubernetesRuntimeFactoryConfig.setJobNamespace("test_jobNamespace");
         kubernetesRuntimeFactoryConfig.setPulsarDockerImageName("test_dockerImage");
+        kubernetesRuntimeFactoryConfig.setFunctionDockerImages(imageNames);
         kubernetesRuntimeFactoryConfig.setImagePullPolicy("test_imagePullPolicy");
         workerConfig.setFunctionRuntimeFactoryClassName(KubernetesRuntimeFactory.class.getName());
         workerConfig.setFunctionRuntimeFactoryConfigs(

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeTest.java

@@ -185,6 +185,7 @@ KubernetesRuntimeFactory createKubernetesRuntimeFactory(String extraDepsDir, int
         kubernetesRuntimeFactoryConfig.setK8Uri(null);
         kubernetesRuntimeFactoryConfig.setJobNamespace(null);
         kubernetesRuntimeFactoryConfig.setPulsarDockerImageName(null);
+        kubernetesRuntimeFactoryConfig.setFunctionDockerImages(null);
         kubernetesRuntimeFactoryConfig.setImagePullPolicy(null);
         kubernetesRuntimeFactoryConfig.setPulsarRootDir(pulsarRootDir);
         kubernetesRuntimeFactoryConfig.setSubmittingInsidePod(false);

site2/docs/functions-runtime.md

@@ -67,6 +67,14 @@ kubernetesContainerFactory:
   jobNamespace:
   # the docker image to run function instance. by default it is `apachepulsar/pulsar`
   pulsarDockerImageName:
+  # the docker image to run function instance according to different configurations provided by users.
+  # By default it is `apachepulsar/pulsar`.
+  # e.g:
+  # functionDockerImages:
+  #   JAVA: JAVA_IMAGE_NAME
+  #   PYTHON: PYTHON_IMAGE_NAME
+  #   GO: GO_IMAGE_NAME
+  functionDockerImages:
   # the root directory of pulsar home directory in `pulsarDockerImageName`. by default it is `/pulsar`.
   # if you are using your own built image in `pulsarDockerImageName`, you need to set this setting accordingly
   pulsarRootDir: