[improve][fn] Upgrade Kubernetes client-java version to 23.0.0 and add k8s e2e integration tests

.github/workflows/pulsar-ci.yaml

@@ -683,6 +683,9 @@ jobs:
           - name: Upgrade
             group: UPGRADE
 
+          - name: Kubernetes
+            group: PULSAR_K8S
+
     steps:
       - name: checkout
         uses: actions/checkout@v4

build/build_java_test_image.sh

@@ -21,5 +21,5 @@
 SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 cd "$SCRIPT_DIR/.."
 mvn -am -pl tests/docker-images/java-test-image -Pcore-modules,-main,integrationTests,docker \
-  -Dmaven.test.skip=true -DskipSourceReleaseAssembly=true -Dspotbugs.skip=true -Dcheckstyle.skip=true -Dlicense.skip=true \
+  -DskipTests -DskipSourceReleaseAssembly=true -Dspotbugs.skip=true -Dcheckstyle.skip=true -Dlicense.skip=true \
   "$@" install

build/run_integration_group.sh

@@ -207,6 +207,10 @@ test_group_sql() {
   mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-sql.xml -DintegrationTests -DtestForkCount=1 -DtestReuseFork=false
 }
 
+test_group_pulsar_k8s() {
+  mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-k8s.xml -DintegrationTests
+}
+
 test_group_pulsar_io() {
   mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-io-sources.xml -DintegrationTests -Dgroups=source
   mvn_run_integration_test --skip-build-deps "$@" -DintegrationTestSuiteFile=pulsar-io-sinks.xml -DintegrationTests -Dgroups=sink

distribution/server/src/assemble/LICENSE.bin.txt

@@ -267,7 +267,7 @@ The Apache Software License, Version 2.0
  * Bitbucket -- org.bitbucket.b_c-jose4j-0.9.4.jar
  * Gson
     - com.google.code.gson-gson-2.8.9.jar
-    - io.gsonfire-gson-fire-1.8.5.jar
+    - io.gsonfire-gson-fire-1.9.0.jar
  * Guava
     - com.google.guava-guava-33.4.8-jre.jar
     - com.google.guava-failureaccess-1.0.3.jar
@@ -472,9 +472,9 @@ The Apache Software License, Version 2.0
   * Apache Yetus
     - org.apache.yetus-audience-annotations-0.12.0.jar
   * Kubernetes Client
-    - io.kubernetes-client-java-18.0.0.jar
-    - io.kubernetes-client-java-api-18.0.0.jar
-    - io.kubernetes-client-java-proto-18.0.0.jar
+    - io.kubernetes-client-java-23.0.0.jar
+    - io.kubernetes-client-java-api-23.0.0.jar
+    - io.kubernetes-client-java-proto-23.0.0.jar
   * Dropwizard
     - io.dropwizard.metrics-metrics-core-4.1.12.1.jar
     - io.dropwizard.metrics-metrics-graphite-4.1.12.1.jar

pom.xml

@@ -283,7 +283,7 @@ flexible messaging model and an intuitive client API.</description>
     <jakarta.xml.bind.version>2.3.3</jakarta.xml.bind.version>
     <jakarta.validation.version>2.0.2</jakarta.validation.version>
     <jna.version>5.12.1</jna.version>
-    <kubernetesclient.version>18.0.0</kubernetesclient.version>
+    <kubernetesclient.version>23.0.0</kubernetesclient.version>
     <jose4j.version>0.9.4</jose4j.version>
     <okhttp3.version>5.3.1</okhttp3.version>
     <!-- use okio version that matches the okhttp3 version -->

pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/JwksCache.java

@@ -169,7 +169,7 @@ CompletableFuture<Jwk> getJwkFromKubernetesApiServer(String keyId) {
     private CompletableFuture<List<Jwk>> getJwksFromKubernetesApiServer() {
         CompletableFuture<List<Jwk>> future = new CompletableFuture<>();
         try {
-            openidApi.getServiceAccountIssuerOpenIDKeysetAsync(new ApiCallback<String>() {
+            openidApi.getServiceAccountIssuerOpenIDKeyset().executeAsync(new ApiCallback<String>() {
                 @Override
                 public void onFailure(ApiException e, int statusCode, Map<String, List<String>> responseHeaders) {
                     authenticationProvider.incrementFailureMetric(ERROR_RETRIEVING_PUBLIC_KEY);

pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/OpenIDProviderMetadataCache.java

@@ -168,7 +168,7 @@ CompletableFuture<OpenIDProviderMetadata> getOpenIDProviderMetadataForKubernetes
     private CompletableFuture<OpenIDProviderMetadata> loadOpenIDProviderMetadataForKubernetesApiServer() {
         CompletableFuture<OpenIDProviderMetadata> future = new CompletableFuture<>();
         try {
-            wellKnownApi.getServiceAccountIssuerOpenIDConfigurationAsync(new ApiCallback<>() {
+            wellKnownApi.getServiceAccountIssuerOpenIDConfiguration().executeAsync(new ApiCallback<>() {
                 @Override
                 public void onFailure(ApiException e, int statusCode, Map<String, List<String>> responseHeaders) {
                     authenticationProvider.incrementFailureMetric(ERROR_RETRIEVING_PROVIDER_METADATA);

pulsar-broker-auth-oidc/src/test/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenIDIntegrationTest.java

@@ -117,12 +117,10 @@ void beforeClass() throws IOException {
                                         """.replace("%s", server.baseUrl()))));
 
         // Set up a correct openid-configuration that the k8s integration test can use
-        // NOTE: integration tests revealed that the k8s client adds a trailing slash to the openid-configuration
-        // endpoint.
         // NOTE: the jwks_uri is ignored, so we supply one that would fail here to ensure that we are not implicitly
         // relying on the jwks_uri.
         server.stubFor(
-                get(urlEqualTo("/k8s/.well-known/openid-configuration/"))
+                get(urlEqualTo("/k8s/.well-known/openid-configuration"))
                         .willReturn(aResponse()
                                 .withHeader("Content-Type", "application/json")
                                 .withBody("""
@@ -170,7 +168,7 @@ void beforeClass() throws IOException {
         // Set up JWKS endpoint with a valid and an invalid public key
         // The url matches are for both the normal and the k8s endpoints
         server.stubFor(
-                get(urlMatching("/keys|/k8s/openid/v1/jwks/"))
+                get(urlMatching("/keys|/k8s/openid/v1/jwks"))
                         .willReturn(aResponse()
                                 .withHeader("Content-Type", "application/json")
                                 .withBody(

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/KubernetesSecretsTokenAuthProvider.java

@@ -179,9 +179,10 @@ public void cleanUpAuthData(Function.FunctionDetails funcDetails, Optional<Funct
                         // make sure secretName is not null or empty string.
                         // If deleteNamespacedSecret is called and secret name is null or empty string
                         // it will delete all the secrets in the namespace
-                        coreClient.deleteNamespacedSecret(secretName,
-                                kubeNamespace, null, null,
-                                0, null, "Foreground", null);
+                        coreClient.deleteNamespacedSecret(secretName, kubeNamespace)
+                                .gracePeriodSeconds(0)
+                                .propagationPolicy("Foreground")
+                                .execute();
                     } catch (ApiException e) {
                         // if already deleted
                         if (e.getCode() == HTTP_NOT_FOUND) {
@@ -205,8 +206,7 @@ public void cleanUpAuthData(Function.FunctionDetails funcDetails, Optional<Funct
                 .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
                 .supplier(() -> {
                     try {
-                        coreClient.readNamespacedSecret(secretName, kubeNamespace, null);
-
+                        coreClient.readNamespacedSecret(secretName, kubeNamespace).execute();
                     } catch (ApiException e) {
                         // statefulset is gone
                         if (e.getCode() == HTTP_NOT_FOUND) {
@@ -304,15 +304,12 @@ private void upsertSecret(String token, Function.FunctionDetails funcDetails, St
                             .data(buildSecretMap(token));
 
                     try {
-                        coreClient.createNamespacedSecret(kubeNamespace, v1Secret, null, null, null, null);
+                        coreClient.createNamespacedSecret(kubeNamespace, v1Secret).execute();
                     } catch (ApiException e) {
                         if (e.getCode() == HTTP_CONFLICT) {
                             try {
-                                coreClient
-                                        .replaceNamespacedSecret(secretName, kubeNamespace, v1Secret,
-                                                null, null, null, null);
+                                coreClient.replaceNamespacedSecret(secretName, kubeNamespace, v1Secret).execute();
                                 return Actions.ActionResult.builder().success(true).build();
-
                             } catch (ApiException e1) {
                                 String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                                 return Actions.ActionResult.builder()
@@ -366,7 +363,7 @@ private String createSecret(String token, Function.FunctionDetails funcDetails)
                             .metadata(new V1ObjectMeta().name(getSecretName(id)))
                             .data(buildSecretMap(token));
                     try {
-                        coreClient.createNamespacedSecret(kubeNamespace, v1Secret, null, null, null, null);
+                        coreClient.createNamespacedSecret(kubeNamespace, v1Secret).execute();
                     } catch (ApiException e) {
                         // already exists
                         if (e.getCode() == HTTP_CONFLICT) {