[fix][fn] Supply download auth params when provided for k8s runtime (#…

…20144)

PIP: #19849

### Motivation

The new `KubernetesServiceAccountTokenAuthProvider` introduced by #19888  does not configure the authentication for download because there is an unnecessary check that the `getFunctionAuthenticationSpec` is not `null`. Given that we do not use this spec in creating the auth, it is unnecessary to use here. Further, when we construct the function's authentication, we do not have this null check. See:

https://github.com/apache/pulsar/blob/ec102fb024a6ea2b195826778300f20e330dff06/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java#L411-L417

### Modifications

* Update the `KubernetesRuntime` to add authentication params to the download command when provided.

### Verifying this change

A new test is added.

### Documentation

- [x] `doc-not-needed`

### Matching PR in forked repository

PR in forked repository: Skipping for this trivial PR

(cherry picked from commit 4f2b8e2)

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

@@ -879,8 +879,7 @@ private List<String> getDownloadCommand(String tenant, String namespace, String
         // add auth plugin and parameters if necessary
         if (authenticationEnabled && authConfig != null) {
             if (isNotBlank(authConfig.getClientAuthenticationPlugin())
-                    && isNotBlank(authConfig.getClientAuthenticationParameters())
-                    && instanceConfig.getFunctionAuthenticationSpec() != null) {
+                    && isNotBlank(authConfig.getClientAuthenticationParameters())) {
                 cmd.addAll(Arrays.asList(
                         "--auth-plugin",
                         authConfig.getClientAuthenticationPlugin(),

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

@@ -861,6 +861,32 @@ public void testCustomKubernetesDownloadCommandsWithAuth() throws Exception {
         assertTrue(containerCommand.contains(expectedDownloadCommand), "Found:" + containerCommand);
     }
 
+    @Test
+    public void testCustomKubernetesDownloadCommandsWithAuthWithoutAuthSpec() throws Exception {
+        InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
+        config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false));
+
+        factory = createKubernetesRuntimeFactory(null,
+                10, 1.0, 1.0, Optional.empty(), null, wconfig -> {
+                    wconfig.setAuthenticationEnabled(true);
+                }, AuthenticationConfig.builder()
+                        .clientAuthenticationPlugin("com.MyAuth")
+                        .clientAuthenticationParameters("{\"authParam1\": \"authParamValue1\"}")
+                        .build());
+
+        KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, null, null, 30l);
+        V1StatefulSet spec = container.createStatefulSet();
+        String expectedDownloadCommand = "pulsar-admin --admin-url " + pulsarAdminUrl
+                + " --auth-plugin com.MyAuth --auth-params {\"authParam1\": \"authParamValue1\"}"
+                + " functions download "
+                + "--tenant " + TEST_TENANT
+                + " --namespace " + TEST_NAMESPACE
+                + " --name " + TEST_NAME
+                + " --destination-file " + pulsarRootDir + "/" + userJarFile;
+        String containerCommand = spec.getSpec().getTemplate().getSpec().getContainers().get(0).getCommand().get(2);
+        assertTrue(containerCommand.contains(expectedDownloadCommand), "Found:" + containerCommand);
+    }
+
     InstanceConfig createGolangInstanceConfig() {
         InstanceConfig config = new InstanceConfig();