[Pulsar SQL] Fix Pulsar SQL CI test (#7898)

### Motivation

Currently, the Pulsar SQL CI test is blocked. 

### Modifications

1. Build a new docker image before CI test to make sure the latest changes are covered.
2. Remove the config `distributed-joins-enabled` in the `$PULSAR_HOME/conf/presto/config.properties`, this config was discarded in presto version `332`.
3. Add the config `-Dpresto-temporarily-allow-java8=true` in `$PULSAR_HOME/conf/presto/jvm.config`, this config is needed if use Java 8 environment.
4. The ClientBuilder will be loaded by two different ClassLoaders, in the Pulsar SQL CI test the ClientBuilder will be created by unexpected ClassLoader, we need to specify the ClassLoader for it.

.github/workflows/ci-integration-sql.yaml

@@ -73,15 +73,14 @@ jobs:
         if: steps.docs.outputs.changed_only == 'no'
         run: mvn -q -B -ntp clean install -DskipTests
 
-#      Flaky Test: https://github.com/apache/pulsar/issues/7750
-#      - name: build pulsar image
-#        if: steps.docs.outputs.changed_only == 'no'
-#        run: mvn -B -f docker/pulsar/pom.xml install -am -Pdocker -DskipTests -Ddocker.nocache=true
-#
-#      - name: build pulsar-all image
-#        if: steps.docs.outputs.changed_only == 'no'
-#        run: mvn -B -f docker/pulsar-all/pom.xml install -am -Pdocker -DskipTests -Ddocker.nocache=true
-#
+      - name: build pulsar image
+        if: steps.docs.outputs.changed_only == 'no'
+        run: mvn -B -f docker/pulsar/pom.xml install -am -Pdocker -DskipTests -Ddocker.nocache=true
+
+      - name: build pulsar-all image
+        if: steps.docs.outputs.changed_only == 'no'
+        run: mvn -B -f docker/pulsar-all/pom.xml install -am -Pdocker -DskipTests -Ddocker.nocache=true
+
       - name: build artifacts and docker pulsar latest test image
         if: steps.docs.outputs.changed_only == 'no'
         run: mvn -B -f tests/docker-images/pom.xml install -am -Pdocker -DskipTests

conf/presto/config.properties

@@ -39,6 +39,4 @@ query.min-expire-age=30m
 
 presto.version=testversion
 
-distributed-joins-enabled=true
-
 node-scheduler.include-coordinator=true

conf/presto/jvm.config

@@ -25,3 +25,4 @@
 -XX:+ExplicitGCInvokesConcurrent
 -XX:+HeapDumpOnOutOfMemoryError
 -XX:+ExitOnOutOfMemoryError
+-Dpresto-temporarily-allow-java8=true

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java

@@ -132,7 +132,7 @@ public class PulsarAdmin implements Closeable {
     public PulsarAdmin(String serviceUrl, ClientConfigurationData clientConfigData) throws PulsarClientException {
         this(serviceUrl, clientConfigData, DEFAULT_CONNECT_TIMEOUT_SECONDS, TimeUnit.SECONDS,
                 DEFAULT_READ_TIMEOUT_SECONDS, TimeUnit.SECONDS,
-                DEFAULT_REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+                DEFAULT_REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS, null);
 
     }
 
@@ -143,7 +143,8 @@ public class PulsarAdmin implements Closeable {
                        int readTimeout,
                        TimeUnit readTimeoutUnit,
                        int requestTimeout,
-                       TimeUnit requestTimeoutUnit) throws PulsarClientException {
+                       TimeUnit requestTimeoutUnit,
+                       ClassLoader clientBuilderClassLoader) throws PulsarClientException {
         this.connectTimeout = connectTimeout;
         this.connectTimeoutUnit = connectTimeoutUnit;
         this.readTimeout = readTimeout;
@@ -171,6 +172,12 @@ public class PulsarAdmin implements Closeable {
         httpConfig.register(MultiPartFeature.class);
         httpConfig.connectorProvider(asyncConnectorProvider);
 
+        ClassLoader originalCtxLoader = null;
+        if (clientBuilderClassLoader != null) {
+            originalCtxLoader = Thread.currentThread().getContextClassLoader();
+            Thread.currentThread().setContextClassLoader(clientBuilderClassLoader);
+        }
+
         ClientBuilder clientBuilder = ClientBuilder.newBuilder()
                 .withConfig(httpConfig)
                 .connectTimeout(this.connectTimeout, this.connectTimeoutUnit)
@@ -207,6 +214,10 @@ public class PulsarAdmin implements Closeable {
         this.worker = new WorkerImpl(root, auth, readTimeoutMs);
         this.schemas = new SchemasImpl(root, auth, readTimeoutMs);
         this.bookies = new BookiesImpl(root, auth, readTimeoutMs);
+
+        if (originalCtxLoader != null) {
+            Thread.currentThread().setContextClassLoader(originalCtxLoader);
+        }
     }
 
     /**

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java

@@ -252,4 +252,10 @@ public interface PulsarAdminBuilder {
      */
     PulsarAdminBuilder requestTimeout(int requestTimeout, TimeUnit requestTimeoutUnit);
 
+    /**
+     *
+     * @return
+     */
+    PulsarAdminBuilder setContextClassLoader(ClassLoader clientBuilderClassLoader);
+
 }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java

@@ -39,12 +39,13 @@ public class PulsarAdminBuilderImpl implements PulsarAdminBuilder {
     private TimeUnit connectTimeoutUnit = TimeUnit.SECONDS;
     private TimeUnit readTimeoutUnit = TimeUnit.SECONDS;
     private TimeUnit requestTimeoutUnit = TimeUnit.SECONDS;
+    private ClassLoader clientBuilderClassLoader = null;
 
     @Override
     public PulsarAdmin build() throws PulsarClientException {
         return new PulsarAdmin(conf.getServiceUrl(),
                 conf, connectTimeout, connectTimeoutUnit, readTimeout, readTimeoutUnit,
-                requestTimeout, requestTimeoutUnit);
+                requestTimeout, requestTimeoutUnit, clientBuilderClassLoader);
     }
 
     public PulsarAdminBuilderImpl() {
@@ -166,4 +167,10 @@ public class PulsarAdminBuilderImpl implements PulsarAdminBuilder {
         this.requestTimeoutUnit = requestTimeoutUnit;
         return this;
     }
+
+    @Override
+    public PulsarAdminBuilder setContextClassLoader(ClassLoader clientBuilderClassLoader) {
+        this.clientBuilderClassLoader = clientBuilderClassLoader;
+        return this;
+    }
 }

pulsar-sql/presto-distribution/LICENSE

@@ -251,6 +251,7 @@ The Apache Software License, Version 2.0
     - netty-transport-native-unix-common-4.1.51.Final-linux-x86_64.jar
  * Joda Time
     - joda-time-2.10.5.jar
+    - joda-time-2.10.1.jar
  * Jetty
     - http2-client-9.4.27.v20200227.jar
     - http2-common-9.4.27.v20200227.jar
@@ -546,7 +547,6 @@ CDDL-1.1 -- licenses/LICENSE-CDDL-1.1.txt
 
 Eclipse Public License - v2.0 -- licenses/LICENSE-EPL-2.0.txt
  * jakarta.annotation-api-1.3.5.jar
- * jakarta.ws.rs-api-2.1.6.jar
 
 Public Domain (CC0) -- licenses/LICENSE-CC0.txt
  * HdrHistogram

pulsar-sql/presto-pulsar-plugin/src/assembly/assembly.xml

@@ -31,6 +31,9 @@
             <outputDirectory>/</outputDirectory>
             <useProjectArtifact>false</useProjectArtifact>
             <scope>runtime</scope>
+            <excludes>
+              <exclude>jakarta.ws.rs:jakarta.ws.rs-api</exclude>
+            </excludes>
         </dependencySet>
     </dependencySets>
 </assembly>
\ No newline at end of file

pulsar-sql/presto-pulsar/pom.xml

@@ -95,6 +95,12 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+          <groupId>joda-time</groupId>
+          <artifactId>joda-time</artifactId>
+          <version>${joda.version}</version>
+        </dependency>
+
     </dependencies>
 
     <build>

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java

@@ -26,6 +26,8 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.regex.Matcher;
 import javax.validation.constraints.NotNull;
+import javax.ws.rs.client.ClientBuilder;
+
 import org.apache.bookkeeper.stats.NullStatsProvider;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
@@ -396,6 +398,7 @@ public class PulsarConnectorConfig implements AutoCloseable {
                 builder.tlsTrustCertsFilePath(getTlsTrustCertsFilePath());
             }
 
+            builder.setContextClassLoader(ClientBuilder.class.getClassLoader());
             this.pulsarAdmin = builder.serviceHttpUrl(getBrokerServiceUrl()).build();
         }
         return this.pulsarAdmin;