diff --git a/.github/workflows/ci-go-functions.yaml b/.github/workflows/ci-go-functions.yaml index 655503849b1c3..14e4c87461535 100644 --- a/.github/workflows/ci-go-functions.yaml +++ b/.github/workflows/ci-go-functions.yaml @@ -32,7 +32,7 @@ concurrency: cancel-in-progress: true env: - MAVEN_OPTS: -Xss1500k -Xmx1024m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 + MAVEN_OPTS: -Xss1500k -Xmx2048m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 jobs: preconditions: diff --git a/.github/workflows/ci-maven-cache-update.yaml b/.github/workflows/ci-maven-cache-update.yaml index 53dff03c248cc..ebead98e1ae28 100644 --- a/.github/workflows/ci-maven-cache-update.yaml +++ b/.github/workflows/ci-maven-cache-update.yaml @@ -42,7 +42,7 @@ on: - cron: '30 */12 * * *' env: - MAVEN_OPTS: -Xss1500k -Xmx1024m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 + MAVEN_OPTS: -Xss1500k -Xmx2048m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 JDK_DISTRIBUTION: corretto jobs: diff --git a/.github/workflows/ci-owasp-dependency-check.yaml b/.github/workflows/ci-owasp-dependency-check.yaml index a273e902c88d2..278f5594d6731 100644 --- a/.github/workflows/ci-owasp-dependency-check.yaml +++ b/.github/workflows/ci-owasp-dependency-check.yaml @@ -24,7 +24,7 @@ on: workflow_dispatch: env: - MAVEN_OPTS: -Xss1500k -Xmx1024m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 + MAVEN_OPTS: -Xss1500k -Xmx2048m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 JDK_DISTRIBUTION: corretto jobs: diff --git a/.github/workflows/pulsar-ci-flaky.yaml b/.github/workflows/pulsar-ci-flaky.yaml index a92e5cd26c35b..ddb2bdcd39caa 100644 --- a/.github/workflows/pulsar-ci-flaky.yaml +++ b/.github/workflows/pulsar-ci-flaky.yaml @@ -65,7 +65,7 @@ concurrency: cancel-in-progress: true env: - MAVEN_OPTS: -Xss1500k -Xmx1500m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 + MAVEN_OPTS: -Xss1500k -Xmx2048m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 # defines the retention period for the intermediate build artifacts needed for rerunning a failed build job # it's possible to rerun individual failed jobs when the build artifacts are available # if the artifacts have already been expired, the complete workflow can be rerun by closing and reopening the PR or by rebasing the PR diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml index 1160a0d1ec363..2f36cdbb29471 100644 --- a/.github/workflows/pulsar-ci.yaml +++ b/.github/workflows/pulsar-ci.yaml @@ -65,7 +65,7 @@ concurrency: cancel-in-progress: true env: - MAVEN_OPTS: -Xss1500k -Xmx1500m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 + MAVEN_OPTS: -Xss1500k -Xmx2048m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 # defines the retention period for the intermediate build artifacts needed for rerunning a failed build job # it's possible to rerun individual failed jobs when the build artifacts are available # if the artifacts have already been expired, the complete workflow can be rerun by closing and reopening the PR or by rebasing the PR @@ -1493,7 +1493,7 @@ jobs: - name: trigger dependency check run: | mvn -B -ntp verify -PskipDocker,skip-all,owasp-dependency-check -Dcheckstyle.skip=true -DskipTests \ - -pl '!distribution/server,!distribution/io,!distribution/offloaders,!tiered-storage/file-system,!pulsar-io/flume,!pulsar-io/hbase,!pulsar-io/hdfs2,!pulsar-io/hdfs3,!pulsar-io/docs,!pulsar-io/jdbc/openmldb' + -pl '!distribution/server,!distribution/io,!distribution/offloaders,!tiered-storage/file-system,!pulsar-io/flume,!pulsar-io/hbase,!pulsar-io/hdfs3,!pulsar-io/docs,!pulsar-io/jdbc/openmldb' - name: Upload report uses: actions/upload-artifact@v4 diff --git a/deployment/terraform-ansible/deploy-pulsar.yaml b/deployment/terraform-ansible/deploy-pulsar.yaml index db2fd1257ca41..3a9f0fd942c17 100644 --- a/deployment/terraform-ansible/deploy-pulsar.yaml +++ b/deployment/terraform-ansible/deploy-pulsar.yaml @@ -147,7 +147,6 @@ # - file # - flume # - hbase -# - hdfs2 # - hdfs3 # - influxdb # - jdbc-clickhouse diff --git a/distribution/io/src/assemble/io.xml b/distribution/io/src/assemble/io.xml index f98ee14bb20c9..cf7731b4c85ab 100644 --- a/distribution/io/src/assemble/io.xml +++ b/distribution/io/src/assemble/io.xml @@ -63,7 +63,6 @@ ${basedir}/../../pulsar-io/kafka-connect-adaptor-nar/target/pulsar-io-kafka-connect-adaptor-${project.version}.nar ${basedir}/../../pulsar-io/hbase/target/pulsar-io-hbase-${project.version}.nar ${basedir}/../../pulsar-io/kinesis/target/pulsar-io-kinesis-${project.version}.nar - ${basedir}/../../pulsar-io/hdfs2/target/pulsar-io-hdfs2-${project.version}.nar ${basedir}/../../pulsar-io/hdfs3/target/pulsar-io-hdfs3-${project.version}.nar ${basedir}/../../pulsar-io/file/target/pulsar-io-file-${project.version}.nar ${basedir}/../../pulsar-io/data-generator/target/pulsar-io-data-generator-${project.version}.nar diff --git a/distribution/licenses/LICENSE-ASM.txt b/distribution/licenses/LICENSE-ASM.txt new file mode 100644 index 0000000000000..4d191851af43e --- /dev/null +++ b/distribution/licenses/LICENSE-ASM.txt @@ -0,0 +1,28 @@ + + ASM: a very small and fast Java bytecode manipulation framework + Copyright (c) 2000-2011 INRIA, France Telecom + All rights reserved. + + Redistribution and use in source and binary forms, with or without + modification, are permitted provided that the following conditions + are met: + 1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + 2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. + 3. Neither the name of the copyright holders nor the names of its + contributors may be used to endorse or promote products derived from + this software without specific prior written permission. + + THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF + THE POSSIBILITY OF SUCH DAMAGE. diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt index 9f5209bc7fd8d..a33e2d761957f 100644 --- a/distribution/server/src/assemble/LICENSE.bin.txt +++ b/distribution/server/src/assemble/LICENSE.bin.txt @@ -330,7 +330,6 @@ The Apache Software License, Version 2.0 - io.prometheus-simpleclient_common-0.16.0.jar - io.prometheus-simpleclient_hotspot-0.16.0.jar - io.prometheus-simpleclient_httpserver-0.16.0.jar - - io.prometheus-simpleclient_jetty-0.16.0.jar - io.prometheus-simpleclient_log4j2-0.16.0.jar - io.prometheus-simpleclient_servlet-0.16.0.jar - io.prometheus-simpleclient_servlet_common-0.16.0.jar @@ -353,8 +352,8 @@ The Apache Software License, Version 2.0 - org.apache.logging.log4j-log4j-slf4j2-impl-2.23.1.jar - org.apache.logging.log4j-log4j-web-2.23.1.jar * Java Native Access JNA - - net.java.dev.jna-jna-jpms-5.12.1.jar - - net.java.dev.jna-jna-platform-jpms-5.12.1.jar + - net.java.dev.jna-jna-jpms-5.14.0.jar + - net.java.dev.jna-jna-platform-jpms-5.14.0.jar * BookKeeper - org.apache.bookkeeper-bookkeeper-common-4.17.0.jar - org.apache.bookkeeper-bookkeeper-common-allocator-4.17.0.jar @@ -393,25 +392,42 @@ The Apache Software License, Version 2.0 - org.asynchttpclient-async-http-client-2.12.1.jar - org.asynchttpclient-async-http-client-netty-utils-2.12.1.jar * Jetty - - org.eclipse.jetty-jetty-client-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-continuation-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-http-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-io-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-proxy-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-security-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-servlets-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-util-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-util-ajax-9.4.54.v20240208.jar - - org.eclipse.jetty.websocket-javax-websocket-client-impl-9.4.54.v20240208.jar - - org.eclipse.jetty.websocket-websocket-api-9.4.54.v20240208.jar - - org.eclipse.jetty.websocket-websocket-client-9.4.54.v20240208.jar - - org.eclipse.jetty.websocket-websocket-common-9.4.54.v20240208.jar - - org.eclipse.jetty.websocket-websocket-server-9.4.54.v20240208.jar - - org.eclipse.jetty.websocket-websocket-servlet-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-alpn-conscrypt-server-9.4.54.v20240208.jar - - org.eclipse.jetty-jetty-alpn-server-9.4.54.v20240208.jar + - org.eclipse.jetty-jetty-alpn-client-12.0.10.jar + - org.eclipse.jetty-jetty-alpn-conscrypt-server-12.0.10.jar + - org.eclipse.jetty-jetty-alpn-server-12.0.10.jar + - org.eclipse.jetty-jetty-client-12.0.10.jar + - org.eclipse.jetty-jetty-ee-12.0.10.jar + - org.eclipse.jetty-jetty-http-12.0.10.jar + - org.eclipse.jetty-jetty-io-12.0.10.jar + - org.eclipse.jetty-jetty-jndi-12.0.10.jar + - org.eclipse.jetty-jetty-plus-12.0.10.jar + - org.eclipse.jetty-jetty-security-12.0.10.jar + - org.eclipse.jetty-jetty-server-12.0.10.jar + - org.eclipse.jetty-jetty-session-12.0.10.jar + - org.eclipse.jetty-jetty-util-12.0.10.jar + - org.eclipse.jetty-jetty-xml-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-annotations-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-nested-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-plus-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-proxy-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-security-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-servlet-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-servlets-12.0.10.jar + - org.eclipse.jetty.ee8-jetty-ee8-webapp-12.0.10.jar + - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-javax-client-12.0.10.jar + - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-javax-common-12.0.10.jar + - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-api-12.0.10.jar + - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-common-12.0.10.jar + - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-server-12.0.10.jar + - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-servlet-12.0.10.jar + - org.eclipse.jetty.toolchain-jetty-javax-websocket-api-1.1.2.jar + - org.eclipse.jetty.toolchain-jetty-servlet-api-4.0.6.jar + - org.eclipse.jetty.websocket-jetty-websocket-core-client-12.0.10.jar + - org.eclipse.jetty.websocket-jetty-websocket-core-common-12.0.10.jar + - org.eclipse.jetty.websocket-jetty-websocket-core-server-12.0.10.jar + - org.eclipse.jetty.websocket-jetty-websocket-jetty-api-12.0.10.jar + - org.eclipse.jetty.websocket-jetty-websocket-jetty-client-12.0.10.jar + - org.eclipse.jetty.websocket-jetty-websocket-jetty-common-12.0.10.jar * SnakeYaml -- org.yaml-snakeyaml-2.0.jar * RocksDB - org.rocksdb-rocksdbjni-7.9.2.jar * Google Error Prone Annotations - com.google.errorprone-error_prone_annotations-2.24.0.jar @@ -546,6 +562,10 @@ BSD 3-clause "New" or "Revised" License * JSR305 -- com.google.code.findbugs-jsr305-3.0.2.jar -- ../licenses/LICENSE-JSR305.txt * JLine -- jline-jline-2.14.6.jar -- ../licenses/LICENSE-JLine.txt * JLine3 -- org.jline-jline-3.21.0.jar -- ../licenses/LICENSE-JLine.txt + * OW2 ASM + - org.ow2.asm-asm-9.7.jar -- ../licenses/LICENSE-ASM.txt + - org.ow2.asm-asm-commons-9.7.jar -- ../licenses/LICENSE-ASM.txt + - org.ow2.asm-asm-tree-9.7.jar -- ../licenses/LICENSE-ASM.txt BSD 2-Clause License * HdrHistogram -- org.hdrhistogram-HdrHistogram-2.1.9.jar -- ../licenses/LICENSE-HdrHistogram.txt @@ -572,7 +592,6 @@ CDDL-1.1 -- ../licenses/LICENSE-CDDL-1.1.txt - com.sun.activation-javax.activation-1.2.0.jar - javax.xml.bind-jaxb-api-2.3.1.jar * Java Servlet API -- javax.servlet-javax.servlet-api-3.1.0.jar - * WebSocket Server API -- javax.websocket-javax.websocket-client-api-1.0.jar * Java Web Service REST API -- javax.ws.rs-javax.ws.rs-api-2.1.jar * HK2 - Dependency Injection Kernel - org.glassfish.hk2-hk2-api-2.6.1.jar @@ -601,6 +620,7 @@ Eclipse Public License - v2.0 -- ../licenses/LICENSE-EPL-2.0.txt * Jakarta Annotations API -- jakarta.annotation-jakarta.annotation-api-1.3.5.jar * Jakarta RESTful Web Services -- jakarta.ws.rs-jakarta.ws.rs-api-2.1.6.jar * Jakarta Injection -- org.glassfish.hk2.external-jakarta.inject-2.6.1.jar + * Jakarta Transactions API -- jakarta.transaction-jakarta.transaction-api-1.3.3.jar Public Domain (CC0) -- ../licenses/LICENSE-CC0.txt * Reactive Streams -- org.reactivestreams-reactive-streams-1.0.3.jar diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt index ff590023ff3a5..82a531ef0160c 100644 --- a/distribution/shell/src/assemble/LICENSE.bin.txt +++ b/distribution/shell/src/assemble/LICENSE.bin.txt @@ -403,14 +403,20 @@ The Apache Software License, Version 2.0 - async-http-client-2.12.1.jar - async-http-client-netty-utils-2.12.1.jar * Jetty - - jetty-client-9.4.54.v20240208.jar - - jetty-http-9.4.54.v20240208.jar - - jetty-io-9.4.54.v20240208.jar - - jetty-util-9.4.54.v20240208.jar - - javax-websocket-client-impl-9.4.54.v20240208.jar - - websocket-api-9.4.54.v20240208.jar - - websocket-client-9.4.54.v20240208.jar - - websocket-common-9.4.54.v20240208.jar + - jetty-alpn-client-12.0.10.jar + - jetty-client-12.0.10.jar + - jetty-ee8-websocket-javax-client-12.0.10.jar + - jetty-ee8-websocket-javax-common-12.0.10.jar + - jetty-ee8-websocket-jetty-api-12.0.10.jar + - jetty-http-12.0.10.jar + - jetty-io-12.0.10.jar + - jetty-javax-websocket-api-1.1.2.jar + - jetty-util-12.0.10.jar + - jetty-websocket-core-client-12.0.10.jar + - jetty-websocket-core-common-12.0.10.jar + - jetty-websocket-jetty-api-12.0.10.jar + - jetty-websocket-jetty-client-12.0.10.jar + - jetty-websocket-jetty-common-12.0.10.jar * SnakeYaml -- snakeyaml-2.0.jar * Google Error Prone Annotations - error_prone_annotations-2.24.0.jar * Javassist -- javassist-3.25.0-GA.jar @@ -438,7 +444,6 @@ CDDL-1.1 -- ../licenses/LICENSE-CDDL-1.1.txt - javax.annotation-api-1.3.2.jar - javax.activation-1.2.0.jar - jaxb-api-2.3.1.jar - * WebSocket Server API -- javax.websocket-client-api-1.0.jar * Java Web Service REST API -- javax.ws.rs-api-2.1.jar * HK2 - Dependency Injection Kernel - hk2-api-2.6.1.jar diff --git a/hadoop-hdfs-client-shaded/pom.xml b/hadoop-hdfs-client-shaded/pom.xml new file mode 100644 index 0000000000000..cdc0a946691b6 --- /dev/null +++ b/hadoop-hdfs-client-shaded/pom.xml @@ -0,0 +1,191 @@ + + + + 4.0.0 + + org.apache.pulsar + pulsar + 3.4.0-SNAPSHOT + + + hadoop-hdfs-client-shaded + Apache Pulsar :: Hadoop HDFS client shaded + + + 9.4.54.v20240208 + + + + + org.apache.hadoop + hadoop-common + ${hadoop3.version} + + + log4j + log4j + + + ch.qos.reload4j + reload4j + + + org.slf4j + * + + + + + org.apache.hadoop + hadoop-hdfs-client + ${hadoop3.version} + + + org.apache.avro + avro + + + org.mortbay.jetty + jetty + + + com.sun.jersey + jersey-core + + + com.sun.jersey + jersey-server + + + javax.servlet + servlet-api + + + + + + + + + org.eclipse.jetty + jetty-bom + ${jetty9.version} + pom + import + + + + + + + + org.apache.maven.plugins + maven-shade-plugin + + + ${shadePluginPhase} + + shade + + + true + true + false + + + + org.apache.hadoop:* + org.eclipse.jetty:* + org.eclipse.jetty.websocket:* + com.sun.jersey:* + javax.ws.rs:* + com.github.pjfanning:jersey-json + org.codehaus.jettison:jettison + com.sun.xml.bind:* + javax.xml.bind:* + jakarta.activation:jakarta.activation-api + javax.activation:javax.activation-api + com.github.stephenc.jcip:jcip-annotations + + + + + + org.eclipse.jetty + org.apache.pulsar.hadoop.shade.org.eclipse.jetty + + + com.sun.jersey + org.apache.pulsar.hadoop.shade.com.sun.jersey + + + com.sun.xml.bind + org.apache.pulsar.hadoop.shade.com.sun.xml.bind + + + javax.ws.rs + org.apache.pulsar.hadoop.shade.javax.ws.rs + + + javax.xml.bind + org.apache.pulsar.hadoop.shade.javax.xml.bind + + + jakarta.activation + org.apache.pulsar.hadoop.shade.jakarta.activation + + + jersey + org.apache.pulsar.hadoop.shade.jersey + + + com.github.pjfanning + org.apache.pulsar.hadoop.shade.com.github.pjfanning + + + org.codehaus.jettison + org.apache.pulsar.hadoop.shade.org.codehaus.jettison + + + net.jcip + org.apache.pulsar.hadoop.shade.net.jcip + + + javax.enterprise.inject + org.apache.pulsar.hadoop.shade.javax.enterprise.inject + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index e3b272babb7bb..f8659bd80c37c 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -68,6 +68,7 @@ import java.util.UUID; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; @@ -147,7 +148,6 @@ import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore; import org.awaitility.Awaitility; import org.awaitility.reflect.WhiteboxImpl; -import org.eclipse.jetty.util.BlockingArrayQueue; import org.mockito.Mockito; import org.testng.Assert; import org.testng.annotations.DataProvider; @@ -3222,8 +3222,8 @@ public void testAddEntryResponseTimeout() throws Exception { private static class CollectCtxAddEntryCallback implements AddEntryCallback { - public List addCompleteCtxList = new BlockingArrayQueue<>(); - public List addFailedCtxList = new BlockingArrayQueue<>(); + public List addCompleteCtxList = new CopyOnWriteArrayList<>(); + public List addFailedCtxList = new CopyOnWriteArrayList<>(); @Override public void addComplete(Position position, ByteBuf entryData, Object ctx) { diff --git a/pom.xml b/pom.xml index 62644d38d167c..abab4722df7f3 100644 --- a/pom.xml +++ b/pom.xml @@ -150,7 +150,7 @@ flexible messaging model and an intuitive client API. 5.1.0 4.1.111.Final 0.0.24.Final - 9.4.54.v20240208 + 12.0.10 2.5.2 2.41 1.10.50 @@ -197,7 +197,6 @@ flexible messaging model and an intuitive client API. 0.4.6 2.7.5 0.4.4-hotfix1 - 3.3.5 2.4.10 1.2.4 8.12.1 @@ -208,9 +207,9 @@ flexible messaging model and an intuitive client API. 1.15.16.Final 0.11.1 0.28.0 - 2.10.2 - 3.3.5 - 2.4.16 + 3.4.0 + ${hadoop3.version} + 2.5.8 32.1.2-jre 1.0 0.16.1 @@ -238,7 +237,7 @@ flexible messaging model and an intuitive client API. 1.2.2 2.3.3 2.0.2 - 5.12.1 + 5.14.0 18.0.0 0.9.4 4.9.3 @@ -328,6 +327,45 @@ flexible messaging model and an intuitive client API. + + com.fasterxml.jackson + jackson-bom + ${jackson.version} + pom + import + + + + org.slf4j + slf4j-bom + ${slf4j.version} + pom + import + + + + org.apache.logging.log4j + log4j-bom + ${log4j2.version} + pom + import + + + + io.netty + netty-bom + ${netty.version} + pom + import + + + + io.grpc + grpc-bom + ${grpc.version} + pom + import + org.jline @@ -479,16 +517,12 @@ flexible messaging model and an intuitive client API. * - log4j-slf4j-impl org.apache.logging.log4j + * - slf4j-log4j12 org.slf4j - - - log4j - log4j + * org.jboss.netty @@ -650,7 +684,7 @@ flexible messaging model and an intuitive client API. ${bookkeeper.version} - slf4j-log4j12 + * org.slf4j @@ -660,6 +694,12 @@ flexible messaging model and an intuitive client API. org.apache.bookkeeper.stats prometheus-metrics-provider ${bookkeeper.version} + + + org.eclipse.jetty + jetty-servlet + + @@ -680,18 +720,6 @@ flexible messaging model and an intuitive client API. ${jose4j.version} - - org.eclipse.jetty - jetty-server - ${jetty.version} - - - - org.eclipse.jetty - jetty-alpn-conscrypt-server - ${jetty.version} - - org.conscrypt conscrypt-openjdk-uber @@ -707,9 +735,9 @@ flexible messaging model and an intuitive client API. - io.netty - netty-bom - ${netty.version} + org.eclipse.jetty.ee8 + jetty-ee8 + ${jetty.version} pom import @@ -797,22 +825,6 @@ flexible messaging model and an intuitive client API. ${commons-text.version} - - org.slf4j - slf4j-bom - ${slf4j.version} - pom - import - - - - org.apache.logging.log4j - log4j-bom - ${log4j2.version} - pom - import - - commons-codec commons-codec @@ -820,33 +832,11 @@ flexible messaging model and an intuitive client API. - org.glassfish.jersey.core - jersey-server - ${jersey.version} - - - - org.glassfish.jersey.core - jersey-client - ${jersey.version} - - - - org.glassfish.jersey.inject - jersey-hk2 - ${jersey.version} - - - - org.glassfish.jersey.containers - jersey-container-servlet-core - ${jersey.version} - - - - org.glassfish.jersey.containers - jersey-container-servlet + org.glassfish.jersey + jersey-bom ${jersey.version} + pom + import @@ -868,14 +858,18 @@ flexible messaging model and an intuitive client API. - org.glassfish.jersey.media - jersey-media-multipart - ${jersey.version} + net.java.dev.jna + jna + ${jna.version} - net.java.dev.jna - jna + jna-platform + ${jna.version} + + + net.java.dev.jna + jna-platform-jpms ${jna.version} @@ -901,13 +895,6 @@ flexible messaging model and an intuitive client API. ${docker-java.version} - - com.fasterxml.jackson - jackson-bom - ${jackson.version} - pom - import - org.codehaus.jettison @@ -1036,12 +1023,6 @@ flexible messaging model and an intuitive client API. ${prometheus.version} - - io.prometheus - simpleclient_jetty - ${prometheus.version} - - io.prometheus simpleclient_caffeine @@ -1167,14 +1148,6 @@ flexible messaging model and an intuitive client API. ${zt-zip.version} - - io.grpc - grpc-bom - ${grpc.version} - pom - import - - io.grpc grpc-all @@ -1654,11 +1627,7 @@ flexible messaging model and an intuitive client API. org.slf4j - slf4j-log4j12 - - - log4j - log4j + * com.fasterxml.jackson.core @@ -1669,8 +1638,8 @@ flexible messaging model and an intuitive client API. * - log4j-slf4j-impl org.apache.logging.log4j + * @@ -2442,6 +2411,7 @@ flexible messaging model and an intuitive client API. pulsar-metadata jetcd-core-shaded jclouds-shaded + hadoop-hdfs-client-shaded pulsar-package-management diff --git a/pulsar-bom/pom.xml b/pulsar-bom/pom.xml index 43cc0b205e614..96405fdd97a91 100644 --- a/pulsar-bom/pom.xml +++ b/pulsar-bom/pom.xml @@ -495,11 +495,6 @@ pulsar-io-hbase ${project.version} - - org.apache.pulsar - pulsar-io-hdfs2 - ${project.version} - org.apache.pulsar pulsar-io-hdfs3 diff --git a/pulsar-broker-auth-oidc/pom.xml b/pulsar-broker-auth-oidc/pom.xml index 2d46dc7d98b37..791e79226c74c 100644 --- a/pulsar-broker-auth-oidc/pom.xml +++ b/pulsar-broker-auth-oidc/pom.xml @@ -113,7 +113,7 @@ com.github.tomakehurst - wiremock-jre8 + wiremock-jre8-standalone ${wiremock.version} test diff --git a/pulsar-broker-common/pom.xml b/pulsar-broker-common/pom.xml index 2eba685b1a221..eeb947e47be5d 100644 --- a/pulsar-broker-common/pom.xml +++ b/pulsar-broker-common/pom.xml @@ -44,11 +44,6 @@ guava - - io.prometheus - simpleclient_jetty - - javax.servlet javax.servlet-api @@ -69,6 +64,16 @@ jjwt-jackson + + org.eclipse.jetty + jetty-server + + + + org.eclipse.jetty.ee8 + jetty-ee8-servlet + + org.bouncycastle diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/JettyRequestLogFactory.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/JettyRequestLogFactory.java index fc88647eb49ea..a36f8f24f341c 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/JettyRequestLogFactory.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/JettyRequestLogFactory.java @@ -19,12 +19,14 @@ package org.apache.pulsar.broker.web; import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.util.TimeZone; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; import org.eclipse.jetty.io.Connection; import org.eclipse.jetty.io.EndPoint; +import org.eclipse.jetty.server.ConnectionMetaData; import org.eclipse.jetty.server.Connector; import org.eclipse.jetty.server.CustomRequestLog; import org.eclipse.jetty.server.ProxyConnectionFactory; @@ -33,6 +35,7 @@ import org.eclipse.jetty.server.Response; import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.Slf4jRequestLogWriter; +import org.eclipse.jetty.util.Attributes; import org.eclipse.jetty.util.HostPort; import org.eclipse.jetty.util.component.ContainerLifeCycle; @@ -128,14 +131,18 @@ public void log(Request request, Response response) { delegate.log(request, response); StringBuilder sb = requestLogStringBuilder.get(); sb.append(" [R:"); - sb.append(request.getRemoteHost()); + String remoteAddr = Request.getRemoteAddr(request); + sb.append(remoteAddr); sb.append(':'); - sb.append(request.getRemotePort()); - InetSocketAddress realRemoteAddress = lookupRealAddress(request.getHttpChannel().getRemoteAddress()); + int remotePort = Request.getRemotePort(request); + sb.append(remotePort); + + InetSocketAddress realRemoteAddress = + lookupRealAddress(unwrap(request.getConnectionMetaData()).getRemoteSocketAddress()); if (realRemoteAddress != null) { String realRemoteHost = HostPort.normalizeHost(realRemoteAddress.getHostString()); int realRemotePort = realRemoteAddress.getPort(); - if (!realRemoteHost.equals(request.getRemoteHost()) || realRemotePort != request.getRemotePort()) { + if (!realRemoteHost.equals(remoteAddr) || realRemotePort != remotePort) { sb.append(" via "); sb.append(realRemoteHost); sb.append(':'); @@ -143,23 +150,26 @@ public void log(Request request, Response response) { } } sb.append("]->[L:"); - InetSocketAddress realLocalAddress = lookupRealAddress(request.getHttpChannel().getLocalAddress()); + InetSocketAddress realLocalAddress = lookupRealAddress(unwrap(request.getConnectionMetaData()) + .getLocalSocketAddress()); + String localAddr = Request.getLocalAddr(request); + int localPort = Request.getLocalPort(request); if (realLocalAddress != null) { String realLocalHost = HostPort.normalizeHost(realLocalAddress.getHostString()); int realLocalPort = realLocalAddress.getPort(); sb.append(realLocalHost); sb.append(':'); sb.append(realLocalPort); - if (!realLocalHost.equals(request.getLocalAddr()) || realLocalPort != request.getLocalPort()) { + if (!realLocalHost.equals(localAddr) || realLocalPort != localPort) { sb.append(" dst "); - sb.append(request.getLocalAddr()); + sb.append(localAddr); sb.append(':'); - sb.append(request.getLocalPort()); + sb.append(localPort); } } else { - sb.append(request.getLocalAddr()); + sb.append(localAddr); sb.append(':'); - sb.append(request.getLocalPort()); + sb.append(localPort); } sb.append(']'); try { @@ -169,19 +179,27 @@ public void log(Request request, Response response) { } } - private InetSocketAddress lookupRealAddress(InetSocketAddress socketAddress) { - if (socketAddress == null) { + private ConnectionMetaData unwrap(ConnectionMetaData connectionMetaData) { + if (connectionMetaData instanceof Attributes) { + return (ConnectionMetaData) Attributes.unwrap((Attributes) connectionMetaData); + } + return connectionMetaData; + } + + private InetSocketAddress lookupRealAddress(SocketAddress socketAddress) { + if (socketAddress == null || !(socketAddress instanceof InetSocketAddress)) { return null; } + InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress; if (proxyProtocolRealAddressMapping.isEmpty()) { - return socketAddress; + return inetSocketAddress; } - AddressEntry entry = proxyProtocolRealAddressMapping.get(new AddressKey(socketAddress.getHostString(), - socketAddress.getPort())); + AddressEntry entry = proxyProtocolRealAddressMapping.get(new AddressKey(inetSocketAddress.getHostString(), + inetSocketAddress.getPort())); if (entry != null) { return entry.realAddress; } else { - return socketAddress; + return inetSocketAddress; } } diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServlet.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServlet.java index 1ecf4c4e53943..793c99d8f8cc0 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServlet.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServlet.java @@ -20,7 +20,7 @@ import com.google.common.annotations.Beta; import org.apache.pulsar.common.configuration.PulsarConfiguration; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlet.ServletHolder; /** * The additional servlet interface for support additional servlet. diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java index bc1f25c5af933..ba60e50369de4 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java @@ -24,7 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.common.configuration.PulsarConfiguration; import org.apache.pulsar.common.nar.NarClassLoader; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlet.ServletHolder; /** * An additional servlet with it's classloader. diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/metrics/JettyStatisticsCollector.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/metrics/JettyStatisticsCollector.java new file mode 100644 index 0000000000000..c4facc315153d --- /dev/null +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/metrics/JettyStatisticsCollector.java @@ -0,0 +1,111 @@ +/* + * 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.jetty.metrics; + +import io.prometheus.client.Collector; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.eclipse.jetty.server.handler.StatisticsHandler; + +/** + * Collect Prometheus metrics from jetty's org.eclipse.jetty.server.handler.StatisticsHandler. + * + * This is ported from prometheus client_java 0.16 version of JettyStatisticsCollector which is using Jetty 9.x. + * This supports Jetty 12.x. + */ +public class JettyStatisticsCollector extends Collector { + private final StatisticsHandler statisticsHandler; + private static final List EMPTY_LIST = new ArrayList(); + + public JettyStatisticsCollector(StatisticsHandler statisticsHandler) { + this.statisticsHandler = statisticsHandler; + } + + @Override + public List collect() { + return Arrays.asList( + buildCounter("jetty_requests_total", "Number of requests", statisticsHandler.getRequests()), + buildGauge("jetty_requests_active", "Number of requests currently active", + statisticsHandler.getRequestsActive()), + buildGauge("jetty_requests_active_max", "Maximum number of requests that have been active at once", + statisticsHandler.getRequestsActiveMax()), + buildGauge("jetty_request_time_max_seconds", "Maximum time spent handling requests", + statisticsHandler.getRequestTimeMax() / 1000_000_000.0), + buildCounter("jetty_request_time_seconds_total", "Total time spent in all request handling", + statisticsHandler.getRequestTimeTotal() / 1000_000_000.0), + buildCounter("jetty_dispatched_total", "Number of dispatches", statisticsHandler.getHandleTotal()), + buildGauge("jetty_dispatched_active", "Number of dispatches currently active", + statisticsHandler.getHandleActive()), + buildGauge("jetty_dispatched_active_max", "Maximum number of active dispatches being handled", + statisticsHandler.getHandleActiveMax()), + buildGauge("jetty_dispatched_time_max", "Maximum time spent in dispatch handling", + statisticsHandler.getHandleTimeMax() / 1000_000_000.0), + buildCounter("jetty_dispatched_time_seconds_total", "Total time spent in dispatch handling", + statisticsHandler.getHandleTimeTotal() / 1000_000_000.0), + buildStatusCounter(), + buildGauge("jetty_stats_seconds", "Time in seconds stats have been collected for", + statisticsHandler.getStatisticsDuration().toNanos() / 1000_000_000.0), + buildCounter("jetty_responses_bytes_total", "Total number of bytes across all responses", + statisticsHandler.getBytesRead() + statisticsHandler.getBytesWritten()) + ); + } + + private static MetricFamilySamples buildGauge(String name, String help, double value) { + return new MetricFamilySamples( + name, + Type.GAUGE, + help, + Collections.singletonList(new MetricFamilySamples.Sample(name, EMPTY_LIST, EMPTY_LIST, value))); + } + + private static MetricFamilySamples buildCounter(String name, String help, double value) { + return new MetricFamilySamples( + name, + Type.COUNTER, + help, + Collections.singletonList(new MetricFamilySamples.Sample(name, EMPTY_LIST, EMPTY_LIST, value))); + } + + private MetricFamilySamples buildStatusCounter() { + String name = "jetty_responses_total"; + return new MetricFamilySamples( + name, + Type.COUNTER, + "Number of requests with response status", + Arrays.asList( + buildStatusSample(name, "1xx", statisticsHandler.getResponses1xx()), + buildStatusSample(name, "2xx", statisticsHandler.getResponses2xx()), + buildStatusSample(name, "3xx", statisticsHandler.getResponses3xx()), + buildStatusSample(name, "4xx", statisticsHandler.getResponses4xx()), + buildStatusSample(name, "5xx", statisticsHandler.getResponses5xx()) + ) + ); + } + + private static MetricFamilySamples.Sample buildStatusSample(String name, String status, double value) { + return new MetricFamilySamples.Sample( + name, + Collections.singletonList("code"), + Collections.singletonList(status), + value); + } +} \ No newline at end of file diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/package-info.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/metrics/package-info.java similarity index 95% rename from pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/package-info.java rename to pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/metrics/package-info.java index 464c6db341e8f..53cc77e93ea21 100644 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/package-info.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/metrics/package-info.java @@ -16,4 +16,4 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.io.hdfs2; \ No newline at end of file +package org.apache.pulsar.jetty.metrics; diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/tls/JettySslContextFactory.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/tls/JettySslContextFactory.java index 46a86045995f9..4049431f93d4e 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/tls/JettySslContextFactory.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/jetty/tls/JettySslContextFactory.java @@ -63,14 +63,15 @@ public static SslContextFactory.Server createServerSslContextWithKeystore(String requireTrustedClientCertOnConnect, ciphers, protocols); } - public static SslContextFactory createServerSslContext(String sslProviderString, boolean tlsAllowInsecureConnection, - String tlsTrustCertsFilePath, - String tlsCertificateFilePath, - String tlsKeyFilePath, - boolean tlsRequireTrustedClientCertOnConnect, - Set ciphers, - Set protocols, - long certRefreshInSec) { + public static SslContextFactory.Server createServerSslContext(String sslProviderString, + boolean tlsAllowInsecureConnection, + String tlsTrustCertsFilePath, + String tlsCertificateFilePath, + String tlsKeyFilePath, + boolean tlsRequireTrustedClientCertOnConnect, + Set ciphers, + Set protocols, + long certRefreshInSec) { DefaultSslContextBuilder sslCtxRefresher = new DefaultSslContextBuilder(tlsAllowInsecureConnection, tlsTrustCertsFilePath, tlsCertificateFilePath, tlsKeyFilePath, tlsRequireTrustedClientCertOnConnect, certRefreshInSec, sslProviderString); diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServlet.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServlet.java index 22b8fa5ea35f5..f70973964234f 100644 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServlet.java +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServlet.java @@ -19,8 +19,7 @@ package org.apache.pulsar.broker.web.plugin.servlet; import org.apache.pulsar.common.configuration.PulsarConfiguration; -import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlet.ServletHolder; public class MockAdditionalServlet implements AdditionalServlet { diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/jetty/tls/JettySslContextFactoryTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/jetty/tls/JettySslContextFactoryTest.java index 2f0c8b627d581..75793ab48b601 100644 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/jetty/tls/JettySslContextFactoryTest.java +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/jetty/tls/JettySslContextFactoryTest.java @@ -51,7 +51,7 @@ public void testJettyTlsServerTls() throws Exception { @Cleanup("stop") Server server = new Server(); List connectors = new ArrayList<>(); - SslContextFactory factory = JettySslContextFactory.createServerSslContext( + SslContextFactory.Server factory = JettySslContextFactory.createServerSslContext( null, false, Resources.getResource("ssl/my-ca/ca.pem").getPath(), @@ -85,7 +85,7 @@ public void testJettyTlsServerInvalidTlsProtocol() throws Exception { @Cleanup("stop") Server server = new Server(); List connectors = new ArrayList<>(); - SslContextFactory factory = JettySslContextFactory.createServerSslContext( + SslContextFactory.Server factory = JettySslContextFactory.createServerSslContext( null, false, Resources.getResource("ssl/my-ca/ca.pem").getPath(), @@ -123,7 +123,7 @@ public void testJettyTlsServerInvalidCipher() throws Exception { @Cleanup("stop") Server server = new Server(); List connectors = new ArrayList<>(); - SslContextFactory factory = JettySslContextFactory.createServerSslContext( + SslContextFactory.Server factory = JettySslContextFactory.createServerSslContext( null, false, Resources.getResource("ssl/my-ca/ca.pem").getPath(), diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml index 20117ed21db06..54ef44bdfbd75 100644 --- a/pulsar-broker/pom.xml +++ b/pulsar-broker/pom.xml @@ -170,7 +170,7 @@ com.github.tomakehurst - wiremock-jre8 + wiremock-jre8-standalone ${wiremock.version} test @@ -188,6 +188,17 @@ test + + org.eclipse.jetty.websocket + jetty-websocket-jetty-client + test + + + org.eclipse.jetty.ee8.websocket + jetty-ee8-websocket-jetty-api + test + + io.dropwizard.metrics @@ -248,13 +259,13 @@ - org.eclipse.jetty - jetty-servlet + org.eclipse.jetty.ee8 + jetty-ee8-servlet - org.eclipse.jetty - jetty-servlets + org.eclipse.jetty.ee8 + jetty-ee8-servlets @@ -281,14 +292,12 @@ org.glassfish.jersey.test-framework jersey-test-framework-core test - ${jersey.version} org.glassfish.jersey.test-framework.providers jersey-test-framework-provider-grizzly2 test - ${jersey.version} @@ -349,11 +358,6 @@ simpleclient - - io.prometheus - simpleclient_jetty - - io.prometheus simpleclient_hotspot diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 65dd90f7a1235..de4981daff77f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -188,8 +188,8 @@ import org.apache.pulsar.websocket.WebSocketProducerServlet; import org.apache.pulsar.websocket.WebSocketReaderServlet; import org.apache.pulsar.websocket.WebSocketService; -import org.eclipse.jetty.servlet.ServletHolder; -import org.eclipse.jetty.websocket.servlet.WebSocketServlet; +import org.eclipse.jetty.ee8.servlet.ServletHolder; +import org.eclipse.jetty.ee8.websocket.server.JettyWebSocketServlet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1169,19 +1169,19 @@ private void addWebSocketServiceHandler(WebService webService, this.webSocketService = new WebSocketService(null, config); this.webSocketService.start(); - final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService); + final JettyWebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService); webService.addServlet(WebSocketProducerServlet.SERVLET_PATH, new ServletHolder(producerWebSocketServlet), true, attributeMap); webService.addServlet(WebSocketProducerServlet.SERVLET_PATH_V2, new ServletHolder(producerWebSocketServlet), true, attributeMap); - final WebSocketServlet consumerWebSocketServlet = new WebSocketConsumerServlet(webSocketService); + final JettyWebSocketServlet consumerWebSocketServlet = new WebSocketConsumerServlet(webSocketService); webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH, new ServletHolder(consumerWebSocketServlet), true, attributeMap); webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH_V2, new ServletHolder(consumerWebSocketServlet), true, attributeMap); - final WebSocketServlet readerWebSocketServlet = new WebSocketReaderServlet(webSocketService); + final JettyWebSocketServlet readerWebSocketServlet = new WebSocketReaderServlet(webSocketService); webService.addServlet(WebSocketReaderServlet.SERVLET_PATH, new ServletHolder(readerWebSocketServlet), true, attributeMap); webService.addServlet(WebSocketReaderServlet.SERVLET_PATH_V2, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java index 43514d481dcab..9218a2582e45e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java @@ -35,7 +35,7 @@ import javax.servlet.http.HttpServletResponse; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.PulsarService; -import org.eclipse.jetty.server.HttpOutput; +import org.eclipse.jetty.ee8.nested.HttpOutput; @Slf4j public class PulsarPrometheusMetricsServlet extends PrometheusMetricsServlet { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java index 205e02ed75a2e..adf6123d72cf3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java @@ -19,8 +19,6 @@ package org.apache.pulsar.broker.web; import java.io.IOException; -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; import javax.servlet.ServletResponse; import javax.servlet.http.HttpServletResponse; import javax.ws.rs.core.Response; @@ -28,11 +26,6 @@ import org.apache.pulsar.common.intercept.InterceptException; import org.apache.pulsar.common.policies.data.ErrorData; import org.apache.pulsar.common.util.ObjectMapperFactory; -import org.eclipse.jetty.http.HttpField; -import org.eclipse.jetty.http.HttpFields; -import org.eclipse.jetty.http.HttpHeader; -import org.eclipse.jetty.http.HttpVersion; -import org.eclipse.jetty.http.MetaData; /** * Exception handler for handle exception. @@ -41,29 +34,17 @@ public class ExceptionHandler { public void handle(ServletResponse response, Exception ex) throws IOException { + HttpServletResponse httpServletResponse = (HttpServletResponse) response; if (ex instanceof InterceptException) { - if (response instanceof org.eclipse.jetty.server.Response) { - String errorData = ObjectMapperFactory - .getMapper().writer().writeValueAsString(new ErrorData(ex.getMessage())); - byte[] errorBytes = errorData.getBytes(StandardCharsets.UTF_8); - int errorCode = ((InterceptException) ex).getErrorCode(); - HttpFields httpFields = new HttpFields(); - HttpField httpField = new HttpField(HttpHeader.CONTENT_TYPE, "application/json;charset=utf-8"); - httpFields.add(httpField); - MetaData.Response info = new MetaData.Response(HttpVersion.HTTP_1_1, errorCode, httpFields); - info.setHttpVersion(HttpVersion.HTTP_1_1); - info.setReason(errorData); - info.setStatus(errorCode); - info.setContentLength(errorBytes.length); - ((org.eclipse.jetty.server.Response) response).getHttpChannel().sendResponse(info, - ByteBuffer.wrap(errorBytes), - true); - } else { - ((HttpServletResponse) response).sendError(((InterceptException) ex).getErrorCode(), - ex.getMessage()); - } + byte[] errorBytes = ObjectMapperFactory + .getMapper().writer().writeValueAsBytes(new ErrorData(ex.getMessage())); + int errorCode = ((InterceptException) ex).getErrorCode(); + httpServletResponse.setStatus(errorCode); + httpServletResponse.setContentType("application/json;charset=utf-8"); + httpServletResponse.setContentLength(errorBytes.length); + httpServletResponse.getOutputStream().write(errorBytes); } else { - ((HttpServletResponse) response).sendError(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), + httpServletResponse.sendError(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), ex.getMessage()); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java index bf484d4f41f65..4082b6683cecf 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java @@ -19,7 +19,6 @@ package org.apache.pulsar.broker.web; import io.prometheus.client.CollectorRegistry; -import io.prometheus.client.jetty.JettyStatisticsCollector; import java.util.ArrayList; import java.util.EnumSet; import java.util.List; @@ -30,7 +29,14 @@ import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.jetty.metrics.JettyStatisticsCollector; import org.apache.pulsar.jetty.tls.JettySslContextFactory; +import org.eclipse.jetty.ee8.nested.ContextHandler; +import org.eclipse.jetty.ee8.nested.ResourceHandler; +import org.eclipse.jetty.ee8.servlet.FilterHolder; +import org.eclipse.jetty.ee8.servlet.ServletContextHandler; +import org.eclipse.jetty.ee8.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlets.QoSFilter; import org.eclipse.jetty.server.ConnectionFactory; import org.eclipse.jetty.server.ConnectionLimit; import org.eclipse.jetty.server.ForwardedRequestCustomizer; @@ -43,18 +49,10 @@ import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.ServerConnector; import org.eclipse.jetty.server.SslConnectionFactory; -import org.eclipse.jetty.server.handler.ContextHandler; import org.eclipse.jetty.server.handler.ContextHandlerCollection; import org.eclipse.jetty.server.handler.DefaultHandler; -import org.eclipse.jetty.server.handler.HandlerCollection; -import org.eclipse.jetty.server.handler.RequestLogHandler; -import org.eclipse.jetty.server.handler.ResourceHandler; import org.eclipse.jetty.server.handler.StatisticsHandler; -import org.eclipse.jetty.servlet.FilterHolder; -import org.eclipse.jetty.servlet.ServletContextHandler; -import org.eclipse.jetty.servlet.ServletHolder; -import org.eclipse.jetty.servlets.QoSFilter; -import org.eclipse.jetty.util.resource.Resource; +import org.eclipse.jetty.util.resource.ResourceFactory; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.glassfish.jersey.media.multipart.MultiPartFeature; import org.glassfish.jersey.server.ResourceConfig; @@ -135,7 +133,7 @@ public WebService(PulsarService pulsar) throws PulsarServerException { Optional tlsPort = config.getWebServicePortTls(); if (tlsPort.isPresent()) { try { - SslContextFactory sslCtxFactory; + SslContextFactory.Server sslCtxFactory; if (config.isTlsEnabledWithKeyStore()) { sslCtxFactory = JettySslContextFactory.createServerSslContextWithKeystore( config.getWebServiceTlsProvider(), @@ -297,39 +295,37 @@ public void addServlet(String path, ServletHolder servletHolder, boolean require } filterInitializer.addFilters(servletContextHandler, requiresAuthentication); - handlers.add(servletContextHandler); + handlers.add(servletContextHandler.get()); } public void addStaticResources(String basePath, String resourcePath) { ContextHandler capHandler = new ContextHandler(); capHandler.setContextPath(basePath); ResourceHandler resHandler = new ResourceHandler(); - resHandler.setBaseResource(Resource.newClassPathResource(resourcePath)); + ResourceFactory resourceFactory = ResourceFactory.root(); + resHandler.setBaseResource(resourceFactory.newClassLoaderResource(resourcePath, true)); resHandler.setEtags(true); resHandler.setCacheControl(WebService.HANDLER_CACHE_CONTROL); capHandler.setHandler(resHandler); - handlers.add(capHandler); + handlers.add(capHandler.get()); } public void start() throws PulsarServerException { try { - RequestLogHandler requestLogHandler = new RequestLogHandler(); boolean showDetailedAddresses = pulsar.getConfiguration().getWebServiceLogDetailedAddresses() != null ? pulsar.getConfiguration().getWebServiceLogDetailedAddresses() : (pulsar.getConfiguration().isWebServiceHaProxyProtocolEnabled() || pulsar.getConfiguration().isWebServiceTrustXForwardedFor()); RequestLog requestLogger = JettyRequestLogFactory.createRequestLogger(showDetailedAddresses, server); - requestLogHandler.setRequestLog(requestLogger); - handlers.add(0, new ContextHandlerCollection()); - handlers.add(requestLogHandler); + server.setRequestLog(requestLogger); ContextHandlerCollection contexts = new ContextHandlerCollection(); - contexts.setHandlers(handlers.toArray(new Handler[handlers.size()])); + contexts.setHandlers(handlers); Handler handlerForContexts = GzipHandlerUtil.wrapWithGzipHandler(contexts, pulsar.getConfig().getHttpServerGzipCompressionExcludedPaths()); - HandlerCollection handlerCollection = new HandlerCollection(); - handlerCollection.setHandlers(new Handler[] {handlerForContexts, new DefaultHandler(), requestLogHandler}); + Handler.Collection handlerCollection = new Handler.Sequence(); + handlerCollection.setHandlers(handlerForContexts, new DefaultHandler()); // Metrics handler StatisticsHandler stats = new StatisticsHandler(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/PrometheusMetricsTestUtil.java b/pulsar-broker/src/test/java/org/apache/pulsar/PrometheusMetricsTestUtil.java index 68826372b7bd6..50663af4f4fcd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/PrometheusMetricsTestUtil.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/PrometheusMetricsTestUtil.java @@ -32,7 +32,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator; import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider; -import org.eclipse.jetty.server.HttpOutput; +import org.eclipse.jetty.ee8.nested.HttpOutput; public class PrometheusMetricsTestUtil { public static void generate(PulsarService pulsar, boolean includeTopicMetrics, boolean includeConsumerMetrics, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java index c9432d65fa2e0..15b881a6042c2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java @@ -37,8 +37,8 @@ import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithPulsarService; import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlets; import org.apache.pulsar.common.configuration.PulsarConfiguration; -import org.eclipse.jetty.server.Request; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.ee8.nested.Request; +import org.eclipse.jetty.ee8.servlet.ServletHolder; import org.mockito.Mockito; import org.testng.Assert; import org.testng.annotations.AfterClass; @@ -145,7 +145,7 @@ public ServletConfig getServletConfig() { @Override public void service(ServletRequest servletRequest, ServletResponse servletResponse) throws ServletException, IOException { - log.info("[service] path: {}", ((Request) servletRequest).getOriginalURI()); + log.info("[service] path: {}", ((Request) servletRequest).getHttpURI()); String value = servletRequest.getParameterMap().get(QUERY_PARAM)[0]; ServletOutputStream servletOutputStream = servletResponse.getOutputStream(); servletResponse.setContentLength(value.getBytes().length); @@ -176,7 +176,7 @@ public WithPulsarServiceServlet(PulsarService pulsar) { @Override public void service(ServletRequest servletRequest, ServletResponse servletResponse) throws ServletException, IOException { - log.info("[service] path: {}", ((Request) servletRequest).getOriginalURI()); + log.info("[service] path: {}", ((Request) servletRequest).getHttpURI()); String value = pulsarService == null ? "null" : PulsarService.class.getName(); ServletOutputStream servletOutputStream = servletResponse.getOutputStream(); servletResponse.setContentLength(value.getBytes().length); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java index 27d72f98c2c49..d052249da6c50 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java @@ -29,6 +29,7 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import javax.servlet.ServletContext; import javax.ws.rs.container.AsyncResponse; import javax.ws.rs.container.TimeoutHandler; import org.apache.pulsar.broker.admin.v2.PersistentTopics; @@ -69,7 +70,7 @@ protected void setup() throws Exception { admin.namespaces().createNamespace("prop/ns-abc"); admin.namespaces().setNamespaceReplicationClusters("prop/ns-abc", Set.of("test")); persistentTopics = spy(PersistentTopics.class); - persistentTopics.setServletContext(new MockServletContext()); + persistentTopics.setServletContext(mock(ServletContext.class)); persistentTopics.setPulsar(pulsar); doReturn(false).when(persistentTopics).isRequestHttps(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index 2894903c0d0c1..4916955573aa8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -45,6 +45,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; +import javax.servlet.ServletContext; import javax.ws.rs.container.AsyncResponse; import javax.ws.rs.core.Response; import javax.ws.rs.core.Response.Status; @@ -137,7 +138,7 @@ public void setup() throws Exception { doNothing().when(properties).validateSuperUserAccess(); namespaces = spy(Namespaces.class); - namespaces.setServletContext(new MockServletContext()); + namespaces.setServletContext(mock(ServletContext.class)); namespaces.setPulsar(pulsar); doReturn("test").when(namespaces).clientAppId(); doReturn(Set.of("use", "usw", "usc", "global")).when(namespaces).clusters(); @@ -154,7 +155,7 @@ public void setup() throws Exception { uriField.setAccessible(true); persistentTopics = spy(PersistentTopics.class); - persistentTopics.setServletContext(new MockServletContext()); + persistentTopics.setServletContext(mock(ServletContext.class)); persistentTopics.setPulsar(pulsar); doReturn("test").when(persistentTopics).clientAppId(); doReturn("persistent").when(persistentTopics).domain(); @@ -164,11 +165,11 @@ public void setup() throws Exception { doNothing().when(persistentTopics).validateAdminAccessForTenant("prop-xyz"); resourceQuotas = spy(ResourceQuotas.class); - resourceQuotas.setServletContext(new MockServletContext()); + resourceQuotas.setServletContext(mock(ServletContext.class)); resourceQuotas.setPulsar(pulsar); brokerStats = spy(BrokerStats.class); - brokerStats.setServletContext(new MockServletContext()); + brokerStats.setServletContext(mock(ServletContext.class)); brokerStats.setPulsar(pulsar); doReturn(false).when(persistentTopics).isRequestHttps(); @@ -177,7 +178,7 @@ public void setup() throws Exception { doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopics).clientAuthData(); schemasResource = spy(SchemasResource.class); - schemasResource.setServletContext(new MockServletContext()); + schemasResource.setServletContext(mock(ServletContext.class)); schemasResource.setPulsar(pulsar); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MockServletContext.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MockServletContext.java deleted file mode 100644 index cdd33da09b833..0000000000000 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MockServletContext.java +++ /dev/null @@ -1,281 +0,0 @@ -/* - * 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.broker.admin; - -import java.io.InputStream; -import java.net.MalformedURLException; -import java.net.URL; -import java.util.Enumeration; -import java.util.EventListener; -import java.util.Map; -import java.util.Set; - -import javax.servlet.Filter; -import javax.servlet.FilterRegistration; -import javax.servlet.RequestDispatcher; -import javax.servlet.Servlet; -import javax.servlet.ServletContext; -import javax.servlet.ServletException; -import javax.servlet.ServletRegistration; -import javax.servlet.ServletRegistration.Dynamic; -import javax.servlet.SessionCookieConfig; -import javax.servlet.SessionTrackingMode; -import javax.servlet.descriptor.JspConfigDescriptor; - -import org.eclipse.jetty.util.AttributesMap; - -public class MockServletContext extends AttributesMap implements ServletContext { - - @Override - public String getContextPath() { - return null; - } - - @Override - public ServletContext getContext(String uripath) { - return null; - } - - @Override - public int getMajorVersion() { - return 0; - } - - @Override - public int getMinorVersion() { - return 0; - } - - @Override - public int getEffectiveMajorVersion() { - return 0; - } - - @Override - public int getEffectiveMinorVersion() { - return 0; - } - - @Override - public String getMimeType(String file) { - return null; - } - - @Override - public Set getResourcePaths(String path) { - return null; - } - - @Override - public URL getResource(String path) throws MalformedURLException { - return null; - } - - @Override - public InputStream getResourceAsStream(String path) { - return null; - } - - @Override - public RequestDispatcher getRequestDispatcher(String path) { - return null; - } - - @Override - public RequestDispatcher getNamedDispatcher(String name) { - return null; - } - - @Override - @Deprecated - public Servlet getServlet(String name) throws ServletException { - return null; - } - - @Override - @Deprecated - public Enumeration getServlets() { - return null; - } - - @Override - @Deprecated - public Enumeration getServletNames() { - return null; - } - - @Override - public void log(String msg) { - } - - @Override - @Deprecated - public void log(Exception exception, String msg) { - } - - @Override - public void log(String message, Throwable throwable) { - - } - - @Override - public String getRealPath(String path) { - return null; - } - - @Override - public String getServerInfo() { - return null; - } - - @Override - public String getInitParameter(String name) { - return null; - } - - @Override - public Enumeration getInitParameterNames() { - return null; - } - - @Override - public boolean setInitParameter(String name, String value) { - return false; - } - - @Override - public String getServletContextName() { - return null; - } - - @Override - public Dynamic addServlet(String servletName, String className) { - return null; - } - - @Override - public Dynamic addServlet(String servletName, Servlet servlet) { - return null; - } - - @Override - public Dynamic addServlet(String servletName, Class servletClass) { - return null; - } - - @Override - public T createServlet(Class clazz) throws ServletException { - return null; - } - - @Override - public ServletRegistration getServletRegistration(String servletName) { - return null; - } - - @Override - public Map getServletRegistrations() { - return null; - } - - @Override - public javax.servlet.FilterRegistration.Dynamic addFilter(String filterName, String className) { - return null; - } - - @Override - public javax.servlet.FilterRegistration.Dynamic addFilter(String filterName, Filter filter) { - return null; - } - - @Override - public javax.servlet.FilterRegistration.Dynamic addFilter(String filterName, Class filterClass) { - return null; - } - - @Override - public T createFilter(Class clazz) throws ServletException { - return null; - } - - @Override - public FilterRegistration getFilterRegistration(String filterName) { - return null; - } - - @Override - public Map getFilterRegistrations() { - return null; - } - - @Override - public SessionCookieConfig getSessionCookieConfig() { - return null; - } - - @Override - public void setSessionTrackingModes(Set sessionTrackingModes) { - } - - @Override - public Set getDefaultSessionTrackingModes() { - return null; - } - - @Override - public Set getEffectiveSessionTrackingModes() { - return null; - } - - @Override - public void addListener(String className) { - } - - @Override - public void addListener(T t) { - } - - @Override - public void addListener(Class listenerClass) { - } - - @Override - public T createListener(Class clazz) throws ServletException { - return null; - } - - @Override - public JspConfigDescriptor getJspConfigDescriptor() { - return null; - } - - @Override - public ClassLoader getClassLoader() { - return null; - } - - @Override - public void declareRoles(String... roleNames) { - } - - @Override - public String getVirtualServerName() { - return null; - } -} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java index 1050d9f33b465..166bfa0634367 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java @@ -56,6 +56,7 @@ import java.util.function.Predicate; import java.util.stream.Collectors; import java.util.stream.Stream; +import javax.servlet.ServletContext; import javax.ws.rs.BadRequestException; import javax.ws.rs.ClientErrorException; import javax.ws.rs.WebApplicationException; @@ -208,7 +209,7 @@ private void initAndStartBroker() throws Exception { super.internalSetup(); namespaces = spy(Namespaces.class); - namespaces.setServletContext(new MockServletContext()); + namespaces.setServletContext(mock(ServletContext.class)); namespaces.setPulsar(pulsar); doReturn(false).when(namespaces).isRequestHttps(); doReturn("test").when(namespaces).clientAppId(); @@ -1192,7 +1193,7 @@ public void testValidateTopicOwnership() throws Exception { ownership.set(pulsar.getNamespaceService(), MockOwnershipCache); TopicName topicName = TopicName.get(testNs.getPersistentTopicName("my-topic")); PersistentTopics topics = spy(PersistentTopics.class); - topics.setServletContext(new MockServletContext()); + topics.setServletContext(mock(ServletContext.class)); topics.setPulsar(pulsar); doReturn(false).when(topics).isRequestHttps(); doReturn("test").when(topics).clientAppId(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java index c1e8dfa30994a..155dda7d339e5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.Objects; import java.util.Set; +import javax.servlet.ServletContext; import javax.ws.rs.core.Response; import javax.ws.rs.core.UriInfo; import org.apache.pulsar.broker.admin.v2.Namespaces; @@ -90,7 +91,7 @@ public void setup() throws Exception { super.internalSetup(); namespaces = spy(Namespaces.class); - namespaces.setServletContext(new MockServletContext()); + namespaces.setServletContext(mock(ServletContext.class)); namespaces.setPulsar(pulsar); doReturn(false).when(namespaces).isRequestHttps(); doReturn("test").when(namespaces).clientAppId(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index 55b4c6e1c6f59..f1ee48e717335 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -48,6 +48,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import javax.servlet.ServletContext; import javax.ws.rs.InternalServerErrorException; import javax.ws.rs.WebApplicationException; import javax.ws.rs.container.AsyncResponse; @@ -143,7 +144,7 @@ protected void setup() throws Exception { conf.setTopicLevelPoliciesEnabled(false); super.internalSetup(); persistentTopics = spy(PersistentTopics.class); - persistentTopics.setServletContext(new MockServletContext()); + persistentTopics.setServletContext(mock(ServletContext.class)); persistentTopics.setPulsar(pulsar); doReturn(false).when(persistentTopics).isRequestHttps(); doReturn(null).when(persistentTopics).originalPrincipal(); @@ -153,7 +154,7 @@ protected void setup() throws Exception { doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopics).clientAuthData(); extPersistentTopics = spy(ExtPersistentTopics.class); - extPersistentTopics.setServletContext(new MockServletContext()); + extPersistentTopics.setServletContext(mock(ServletContext.class)); extPersistentTopics.setPulsar(pulsar); doReturn(false).when(extPersistentTopics).isRequestHttps(); doReturn(null).when(extPersistentTopics).originalPrincipal(); @@ -163,7 +164,7 @@ protected void setup() throws Exception { doReturn(mock(AuthenticationDataHttps.class)).when(extPersistentTopics).clientAuthData(); nonPersistentTopic = spy(NonPersistentTopics.class); - nonPersistentTopic.setServletContext(new MockServletContext()); + nonPersistentTopic.setServletContext(mock(ServletContext.class)); nonPersistentTopic.setPulsar(pulsar); namespaceResources = mock(NamespaceResources.class); doReturn(false).when(nonPersistentTopic).isRequestHttps(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java index f2266d3ed5fe3..fc69dc08951f1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java @@ -19,6 +19,7 @@ package org.apache.pulsar.broker.admin; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.fail; @@ -27,6 +28,7 @@ import java.util.Iterator; import java.util.List; import java.util.Set; +import javax.servlet.ServletContext; import org.apache.pulsar.broker.admin.v2.ResourceGroups; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.web.RestException; @@ -51,7 +53,7 @@ public class ResourceGroupsTest extends MockedPulsarServiceBaseTest { protected void setup() throws Exception { super.internalSetup(); resourcegroups = spy(ResourceGroups.class); - resourcegroups.setServletContext(new MockServletContext()); + resourcegroups.setServletContext(mock(ServletContext.class)); resourcegroups.setPulsar(pulsar); doReturn(false).when(resourcegroups).isRequestHttps(); doReturn("test").when(resourcegroups).clientAppId(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java index 91bfa8185e0f0..76e883ac87ed2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java @@ -252,7 +252,11 @@ public void onWebserviceResponse(ServletRequest request, ServletResponse respons } if (response instanceof Response) { Response res = (Response) response; - responseList.add(new ResponseEvent(res.getHttpChannel().getRequest().getRequestURI(), res.getStatus())); + responseList.add(new ResponseEvent(res.getRequest().getHttpURI().asString(), res.getStatus())); + } else if (response instanceof org.eclipse.jetty.ee8.nested.Response) { + org.eclipse.jetty.ee8.nested.Response res = (org.eclipse.jetty.ee8.nested.Response) response; + responseList.add( + new ResponseEvent(res.getHttpChannel().getRequest().getHttpURI().asString(), res.getStatus())); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java index 3c0c2f4e68774..7f5726b66739d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java @@ -23,8 +23,6 @@ import javax.servlet.http.HttpServletResponse; import lombok.SneakyThrows; import org.apache.pulsar.common.intercept.InterceptException; -import org.eclipse.jetty.server.HttpChannel; -import org.eclipse.jetty.server.Response; import org.mockito.Mockito; import org.testng.annotations.Test; @@ -41,6 +39,7 @@ public void testHandle() { String internal = "internal exception"; String illegal = "illegal argument exception "; ExceptionHandler handler = new ExceptionHandler(); + HttpServletResponse response = Mockito.mock(HttpServletResponse.class); handler.handle(response, new InterceptException(PRECONDITION_FAILED_412, restriction)); Mockito.verify(response).sendError(PRECONDITION_FAILED_412, restriction); @@ -50,12 +49,6 @@ public void testHandle() { handler.handle(response, new IllegalArgumentException(illegal)); Mockito.verify(response).sendError(INTERNAL_SERVER_ERROR_500, illegal); - - Response response2 = Mockito.mock(Response.class); - HttpChannel httpChannel = Mockito.mock(HttpChannel.class); - Mockito.when(response2.getHttpChannel()).thenReturn(httpChannel); - handler.handle(response2, new InterceptException(PRECONDITION_FAILED_412, restriction)); - Mockito.verify(httpChannel).sendResponse(Mockito.any(), Mockito.any(), Mockito.anyBoolean()); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceOriginalClientIPTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceOriginalClientIPTest.java index 7f7fa85bd3bb4..68565b1635e4b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceOriginalClientIPTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceOriginalClientIPTest.java @@ -27,9 +27,10 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.assertj.core.api.ThrowingConsumer; import org.awaitility.Awaitility; +import org.eclipse.jetty.client.ContentResponse; import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.ProxyProtocolClientConnectionFactory.V2; -import org.eclipse.jetty.client.api.ContentResponse; +import org.eclipse.jetty.http.HttpField; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -45,7 +46,8 @@ public class WebServiceOriginalClientIPTest extends MockedPulsarServiceBaseTest @Override protected void setup() throws Exception { super.internalSetup(); - httpClient = new HttpClient(new SslContextFactory(true)); + httpClient = new HttpClient(); + httpClient.setSslContextFactory(new SslContextFactory.Client(true)); httpClient.start(); } @@ -81,7 +83,7 @@ public void testClientIPIsPickedFromXForwardedForHeaderAndLogged(boolean tlsEnab performLoggingTest(consoleCaptor -> { // Send a GET request to the metrics URL ContentResponse response = httpClient.newRequest(metricsUrl) - .header("X-Forwarded-For", "11.22.33.44:12345") + .headers(hdrs -> hdrs.ensureField(new HttpField("X-Forwarded-For", "11.22.33.44:12345"))) .send(); // Validate the response @@ -100,7 +102,7 @@ public void testClientIPIsPickedFromForwardedHeaderAndLogged(boolean tlsEnabled) performLoggingTest(consoleCaptor -> { // Send a GET request to the metrics URL ContentResponse response = httpClient.newRequest(metricsUrl) - .header("Forwarded", "for=11.22.33.44:12345") + .headers(hdrs -> hdrs.ensureField(new HttpField("Forwarded", "for=11.22.33.44:12345"))) .send(); // Validate the response diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java index 1e33121259337..86efeb261d543 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java @@ -28,7 +28,7 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.common.configuration.PulsarConfiguration; import org.apache.pulsar.common.nar.NarClassLoader; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlet.ServletHolder; import org.testng.annotations.Test; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java index f65a98fd3f18f..cabcc421590fb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java @@ -20,7 +20,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.common.configuration.PulsarConfiguration; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlet.ServletHolder; public class MockAdditionalServletWithClassLoader implements AdditionalServletWithPulsarService{ @Override diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java index 9ca0bafe00b3b..54aa055da853a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java @@ -20,25 +20,22 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.util.concurrent.ThreadFactoryBuilder; - import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; +import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; - import java.io.IOException; import java.net.InetSocketAddress; import java.util.concurrent.ThreadFactory; import java.util.regex.Pattern; - import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; - import org.apache.pulsar.client.api.MockBrokerServiceHooks.CommandAckHook; import org.apache.pulsar.client.api.MockBrokerServiceHooks.CommandCloseConsumerHook; import org.apache.pulsar.client.api.MockBrokerServiceHooks.CommandCloseProducerHook; @@ -72,9 +69,10 @@ import org.apache.pulsar.common.protocol.PulsarDecoder; import org.apache.pulsar.common.protocol.schema.SchemaVersion; import org.apache.pulsar.common.util.netty.EventLoopUtil; -import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.ee8.nested.AbstractHandler; +import org.eclipse.jetty.ee8.nested.ContextHandler; +import org.eclipse.jetty.ee8.nested.Request; import org.eclipse.jetty.server.Server; -import org.eclipse.jetty.server.handler.AbstractHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -84,7 +82,7 @@ public class MockBrokerService { private LookupData lookupData; - private class genericResponseHandler extends AbstractHandler { + private class GenericResponseHandler extends AbstractHandler { private final ObjectMapper objectMapper = new ObjectMapper(); private final String lookupURI = "/lookup/v2/destination/persistent"; private final String partitionMetadataURI = "/admin/persistent"; @@ -96,7 +94,7 @@ private class genericResponseHandler extends AbstractHandler { private final Pattern multiPartPattern = Pattern.compile(".*/multi-part-.*"); @Override - public void handle(String s, Request baseRequest, HttpServletRequest request, HttpServletResponse response) + public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException { String responseString; log.info("Received HTTP request {}", baseRequest.getRequestURI()); @@ -296,7 +294,7 @@ final protected void handlePong(CommandPong pong) { public MockBrokerService() { server = new Server(0); - server.setHandler(new genericResponseHandler()); + server.setHandler(new ContextHandler("/", new GenericResponseHandler())); } public void start() { @@ -339,7 +337,7 @@ public void startMockBrokerService() throws Exception { @Override public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MaxMessageSize, 0, 4, 0, 4)); - ch.pipeline().addLast("handler", new MockServerCnx()); + ch.pipeline().addLast("handler", (ChannelHandler) new MockServerCnx()); } }); // Bind and start to accept incoming connections. diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssConsumer.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssConsumer.java index ab8372cbfb058..e23a99cf91dc2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssConsumer.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssConsumer.java @@ -34,14 +34,17 @@ import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.websocket.data.ConsumerMessage; import org.eclipse.jetty.websocket.api.Session; -import org.eclipse.jetty.websocket.api.WebSocketAdapter; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen; import org.eclipse.jetty.websocket.api.annotations.WebSocket; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; @Slf4j -@WebSocket(maxTextMessageSize = 64 * 1024) -public class ClientSideEncryptionWssConsumer extends WebSocketAdapter implements Closeable { +@WebSocket +public class ClientSideEncryptionWssConsumer implements Closeable { private Session session; private final CryptoKeyReader cryptoKeyReader; @@ -69,8 +72,8 @@ public ClientSideEncryptionWssConsumer(String webSocketProxyHost, int webSocketP public void start() throws Exception { wssClient = new WebSocketClient(); wssClient.start(); - session = wssClient.connect(this, buildConnectURL(), new ClientUpgradeRequest()).get(); - assertTrue(session.isOpen()); + Session ses = wssClient.connect(this, buildConnectURL(), new ClientUpgradeRequest()).get(); + assertTrue(ses.isOpen()); } private URI buildConnectURL() throws PulsarClientException.CryptoException { @@ -93,24 +96,24 @@ public synchronized ConsumerMessage receive(int timeout, TimeUnit unit) throws E return msg; } - @Override + @OnWebSocketClose public void onWebSocketClose(int statusCode, String reason) { log.info("Connection closed: {} - {}", statusCode, reason); this.session = null; } - @Override + @OnWebSocketOpen public void onWebSocketConnect(Session session) { log.info("Got connect: {}", session); this.session = session; } - @Override + @OnWebSocketError public void onWebSocketError(Throwable cause) { log.error("Received an error", cause); } - @Override + @OnWebSocketMessage public void onWebSocketText(String text) { try { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssProducer.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssProducer.java index f9ae6cd4344e0..d6fcc8b5d256d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssProducer.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ClientSideEncryptionWssProducer.java @@ -18,9 +18,9 @@ */ package org.apache.pulsar.websocket.proxy; -import static org.testng.Assert.assertTrue; import static org.apache.pulsar.common.api.EncryptionContext.EncryptionKey; import static org.apache.pulsar.websocket.proxy.WssClientSideEncryptUtils.EncryptedPayloadAndParam; +import static org.testng.Assert.assertTrue; import java.io.Closeable; import java.io.IOException; import java.net.URI; @@ -42,15 +42,19 @@ import org.apache.pulsar.common.api.proto.MessageIdData; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.websocket.data.ProducerMessage; +import org.eclipse.jetty.websocket.api.Callback; import org.eclipse.jetty.websocket.api.Session; -import org.eclipse.jetty.websocket.api.WebSocketAdapter; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen; import org.eclipse.jetty.websocket.api.annotations.WebSocket; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; @Slf4j -@WebSocket(maxTextMessageSize = 64 * 1024) -public class ClientSideEncryptionWssProducer extends WebSocketAdapter implements Closeable { +@WebSocket +public class ClientSideEncryptionWssProducer implements Closeable { private Session session; private volatile CompletableFuture sendFuture; @@ -80,8 +84,9 @@ public ClientSideEncryptionWssProducer(String webSocketProxyHost, int webSocketP public void start() throws Exception { wssClient = new WebSocketClient(); wssClient.start(); - session = wssClient.connect(this, buildConnectURL(), new ClientUpgradeRequest()).get(); - assertTrue(session.isOpen()); + org.eclipse.jetty.websocket.api.Session ses = + wssClient.connect(this, buildConnectURL(), new ClientUpgradeRequest()).get(); + assertTrue(ses.isOpen()); } private URI buildConnectURL() throws PulsarClientException.CryptoException { @@ -130,7 +135,7 @@ public synchronized MessageIdData sendMessage(ProducerMessage msg) throws Except // Do send. sendFuture = new CompletableFuture<>(); String jsonMsg = ObjectMapperFactory.getMapper().writer().writeValueAsString(msg); - this.session.getRemote().sendString(jsonMsg); + this.session.sendText(jsonMsg, Callback.NOOP); // Wait for response. executor.schedule(() -> { synchronized (ClientSideEncryptionWssProducer.this) { @@ -142,7 +147,7 @@ public synchronized MessageIdData sendMessage(ProducerMessage msg) throws Except return sendFuture.get(); } - @Override + @OnWebSocketClose public void onWebSocketClose(int statusCode, String reason) { log.info("Connection closed: {} - {}", statusCode, reason); this.session = null; @@ -151,18 +156,18 @@ public void onWebSocketClose(int statusCode, String reason) { } } - @Override + @OnWebSocketOpen public void onWebSocketConnect(Session session) { log.info("Got connect: {}", session); this.session = session; } - @Override + @OnWebSocketError public void onWebSocketError(Throwable cause) { log.error("Received an error", cause); } - @Override + @OnWebSocketMessage public void onWebSocketText(String text) { try { ResponseOfSend responseOfSend = diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java index 9ec6a7daf7234..49a9c7d21194e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java @@ -68,7 +68,7 @@ import org.apache.pulsar.websocket.stats.ProxyTopicStat.ProducerStats; import org.awaitility.Awaitility; import org.eclipse.jetty.websocket.api.Session; -import org.eclipse.jetty.websocket.api.UpgradeException; +import org.eclipse.jetty.websocket.api.exceptions.UpgradeException; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; import org.glassfish.jersey.client.ClientConfig; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java index dca4964fc987e..43ff0787983f4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java @@ -24,10 +24,10 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doReturn; import java.net.URI; -import java.security.GeneralSecurityException; import java.util.Optional; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import lombok.Cleanup; import org.apache.pulsar.client.api.TlsProducerConsumerBase; import org.apache.pulsar.client.impl.auth.AuthenticationTls; import org.apache.pulsar.common.util.SecurityUtility; @@ -37,6 +37,7 @@ import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration; import org.apache.pulsar.websocket.service.WebSocketServiceStarter; import org.awaitility.Awaitility; +import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.eclipse.jetty.websocket.api.Session; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; @@ -95,20 +96,25 @@ protected void cleanup() throws Exception { } @Test(timeOut = 30000) - public void socketTest() throws GeneralSecurityException { + public void socketTest() throws Exception { String consumerUri = "wss://localhost:" + proxyServer.getListenPortHTTPS().get() + "/ws/consumer/persistent/my-property/use/my-ns/my-topic/my-sub"; String producerUri = "wss://localhost:" + proxyServer.getListenPortHTTPS().get() + "/ws/producer/persistent/my-property/use/my-ns/my-topic/"; URI consumeUri = URI.create(consumerUri); URI produceUri = URI.create(producerUri); - SslContextFactory sslContextFactory = new SslContextFactory(); + SslContextFactory.Client sslContextFactory = new SslContextFactory.Client(); sslContextFactory.setSslContext(SecurityUtility .createSslContext(false, SecurityUtility.loadCertificatesFromPemFile(CA_CERT_FILE_PATH), null)); + @Cleanup("stop") + HttpClient httpClient = new HttpClient(); + httpClient.setSslContextFactory(sslContextFactory); - WebSocketClient consumeClient = new WebSocketClient(sslContextFactory); + @Cleanup("stop") + WebSocketClient consumeClient = new WebSocketClient(httpClient); SimpleConsumerSocket consumeSocket = new SimpleConsumerSocket(); - WebSocketClient produceClient = new WebSocketClient(sslContextFactory); + @Cleanup("stop") + WebSocketClient produceClient = new WebSocketClient(httpClient); try { consumeClient.start(); @@ -130,9 +136,6 @@ public void socketTest() throws GeneralSecurityException { }); consumeSocket.awaitClose(1, TimeUnit.SECONDS); produceSocket.awaitClose(1, TimeUnit.SECONDS); - } catch (Throwable t) { - log.error(t.getMessage()); - Assert.fail(t.getMessage()); } finally { try { consumeClient.stop(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java index 6510c69d1ae5a..6ea3fa5e19525 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java @@ -29,16 +29,16 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.eclipse.jetty.websocket.api.RemoteEndpoint; +import org.eclipse.jetty.websocket.api.Callback; import org.eclipse.jetty.websocket.api.Session; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen; import org.eclipse.jetty.websocket.api.annotations.WebSocket; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -@WebSocket(maxTextMessageSize = 64 * 1024) +@WebSocket public class SimpleConsumerSocket { private static final String X_PULSAR_MESSAGE_ID = "messageId"; private final CountDownLatch closeLatch; @@ -70,7 +70,7 @@ public void onClose(int statusCode, String reason) { this.closeLatch.countDown(); } - @OnWebSocketConnect + @OnWebSocketOpen public void onConnect(Session session) throws InterruptedException { log.info("Got connect: {}", session); this.session = session; @@ -86,12 +86,12 @@ public synchronized void onMessage(String msg) throws JsonParseException, IOExce String messageId = message.get(X_PULSAR_MESSAGE_ID).getAsString(); consumerBuffer.add(messageId); if (customMessageHandler != null) { - this.getRemote().sendString(customMessageHandler.handle(messageId, message)); + this.getSession().sendText(customMessageHandler.handle(messageId, message), Callback.NOOP); } else { JsonObject ack = new JsonObject(); ack.add("messageId", new JsonPrimitive(messageId)); // Acking the proxy - this.getRemote().sendString(ack.toString()); + this.getSession().sendText(ack.toString(), Callback.NOOP); } } else { consumerBuffer.add(message.toString()); @@ -102,23 +102,19 @@ public void sendPermits(int nbPermits) throws IOException { JsonObject permitMessage = new JsonObject(); permitMessage.add("type", new JsonPrimitive("permit")); permitMessage.add("permitMessages", new JsonPrimitive(nbPermits)); - this.getRemote().sendString(permitMessage.toString()); + this.getSession().sendText(permitMessage.toString(), Callback.NOOP); } public void unsubscribe() throws IOException { JsonObject message = new JsonObject(); message.add("type", new JsonPrimitive("unsubscribe")); - this.getRemote().sendString(message.toString()); + this.getSession().sendText(message.toString(), Callback.NOOP); } public void isEndOfTopic() throws IOException { JsonObject message = new JsonObject(); message.add("type", new JsonPrimitive("isEndOfTopic")); - this.getRemote().sendString(message.toString()); - } - - public RemoteEndpoint getRemote() { - return this.session.getRemote(); + this.getSession().sendText(message.toString(), Callback.NOOP); } public Session getSession() { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java index efd02dc0e5ee0..b3c86c919319c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java @@ -30,16 +30,16 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.websocket.data.ProducerMessage; -import org.eclipse.jetty.websocket.api.RemoteEndpoint; +import org.eclipse.jetty.websocket.api.Callback; import org.eclipse.jetty.websocket.api.Session; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen; import org.eclipse.jetty.websocket.api.annotations.WebSocket; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -@WebSocket(maxTextMessageSize = 64 * 1024) +@WebSocket public class SimpleProducerSocket { private final CountDownLatch closeLatch; @@ -75,7 +75,7 @@ public void onClose(int statusCode, String reason) { this.closeLatch.countDown(); } - @OnWebSocketConnect + @OnWebSocketOpen public void onConnect(Session session) throws Exception { log.info("Got connect: {}", session); this.session = session; @@ -84,7 +84,7 @@ public void onConnect(Session session) throws Exception { public void sendMessage(int totalMsgs) throws Exception { for (int i = 0; i < totalMsgs; i++) { - this.session.getRemote().sendString(getTestJsonPayload(i)); + this.session.sendText(getTestJsonPayload(i), Callback.NOOP); } } @@ -94,10 +94,6 @@ public synchronized void onMessage(String msg) throws JsonParseException { producerBuffer.add(ack.get("messageId").getAsString()); } - public RemoteEndpoint getRemote() { - return this.session.getRemote(); - } - public Session getSession() { return this.session; } diff --git a/pulsar-client-admin/pom.xml b/pulsar-client-admin/pom.xml index bdcbeeef59c3c..b17e07ffbf682 100644 --- a/pulsar-client-admin/pom.xml +++ b/pulsar-client-admin/pom.xml @@ -115,7 +115,7 @@ com.github.tomakehurst - wiremock-jre8 + wiremock-jre8-standalone ${wiremock.version} test diff --git a/pulsar-client-tools/pom.xml b/pulsar-client-tools/pom.xml index 7616b8d3d96f6..f93d58c6599a9 100644 --- a/pulsar-client-tools/pom.xml +++ b/pulsar-client-tools/pom.xml @@ -76,11 +76,6 @@ pulsar-client-messagecrypto-bc ${project.version} - - ${project.groupId} - pulsar-cli-utils - ${project.version} - org.asynchttpclient async-http-client @@ -115,8 +110,19 @@ org.eclipse.jetty.websocket - javax-websocket-client-impl - ${jetty.version} + jetty-websocket-jetty-api + + + org.eclipse.jetty.websocket + jetty-websocket-jetty-client + + + org.eclipse.jetty.ee8.websocket + jetty-ee8-websocket-javax-client + + + org.eclipse.jetty.ee8.websocket + jetty-ee8-websocket-jetty-api io.swagger diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/AbstractCmdConsume.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/AbstractCmdConsume.java index 658b34767b594..053559a595ee0 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/AbstractCmdConsume.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/AbstractCmdConsume.java @@ -42,11 +42,11 @@ import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue; -import org.eclipse.jetty.websocket.api.RemoteEndpoint; +import org.eclipse.jetty.websocket.api.Callback; import org.eclipse.jetty.websocket.api.Session; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen; import org.eclipse.jetty.websocket.api.annotations.WebSocket; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -188,7 +188,7 @@ protected static Map genericRecordToMap(GenericRecord value, boo return res; } - @WebSocket(maxTextMessageSize = 64 * 1024) + @WebSocket public static class ConsumerSocket { private static final String X_PULSAR_MESSAGE_ID = "messageId"; private final CountDownLatch closeLatch; @@ -213,7 +213,7 @@ public void onClose(int statusCode, String reason) { this.closeLatch.countDown(); } - @OnWebSocketConnect + @OnWebSocketOpen public void onConnect(Session session) throws InterruptedException { log.info("Got connect: {}", session); this.session = session; @@ -227,7 +227,7 @@ public synchronized void onMessage(String msg) throws Exception { String messageId = message.get(X_PULSAR_MESSAGE_ID).getAsString(); ack.add("messageId", new JsonPrimitive(messageId)); // Acking the proxy - this.getRemote().sendString(ack.toString()); + this.getSession().sendText(ack.toString(), Callback.NOOP); this.incomingMessages.put(msg); } @@ -235,10 +235,6 @@ public String receive(long timeout, TimeUnit unit) throws Exception { return incomingMessages.poll(timeout, unit); } - public RemoteEndpoint getRemote() { - return this.session.getRemote(); - } - public Session getSession() { return this.session; } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java index 71c172b633713..ab14564a54cd1 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java @@ -38,6 +38,7 @@ import org.apache.pulsar.client.api.SubscriptionMode; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.naming.TopicName; +import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; @@ -247,7 +248,10 @@ private int consumeFromWebSocket(String topic) { URI consumerUri = URI.create(getWebSocketConsumeUri(topic)); - WebSocketClient consumeClient = new WebSocketClient(new SslContextFactory(true)); + HttpClient httpClient = new HttpClient(); + httpClient.setSslContextFactory(new SslContextFactory.Client(true)); + WebSocketClient consumeClient = new WebSocketClient(httpClient); + consumeClient.setMaxTextMessageSize(64 * 1024); ClientUpgradeRequest consumeRequest = new ClientUpgradeRequest(); try { if (authentication != null) { @@ -309,6 +313,17 @@ private int consumeFromWebSocket(String topic) { LOG.info("{} messages successfully consumed", numMessagesConsumed); } + + try { + consumeClient.stop(); + } catch (Exception e) { + LOG.error("Failed to stop websocket-client", e); + } + try { + httpClient.stop(); + } catch (Exception e) { + LOG.error("Failed to stop http-client", e); + } return returnCode; } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdProduce.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdProduce.java index e5a8836602151..82aa8001802a7 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdProduce.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdProduce.java @@ -63,12 +63,13 @@ import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.websocket.data.ProducerMessage; +import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.util.ssl.SslContextFactory; -import org.eclipse.jetty.websocket.api.RemoteEndpoint; +import org.eclipse.jetty.websocket.api.Callback; import org.eclipse.jetty.websocket.api.Session; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect; import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen; import org.eclipse.jetty.websocket.api.annotations.WebSocket; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; @@ -464,7 +465,11 @@ private int publishToWebSocket(String topic) { URI produceUri = URI.create(getWebSocketProduceUri(topic)); - WebSocketClient produceClient = new WebSocketClient(new SslContextFactory(true)); + HttpClient httpClient = new HttpClient(); + httpClient.setSslContextFactory(new SslContextFactory.Client(true)); + WebSocketClient produceClient = new WebSocketClient(httpClient); + produceClient.setMaxTextMessageSize(64 * 1024); + ClientUpgradeRequest produceRequest = new ClientUpgradeRequest(); try { if (authentication != null) { @@ -521,10 +526,21 @@ private int publishToWebSocket(String topic) { LOG.info("{} messages successfully produced", numMessagesSent); } + try { + produceClient.stop(); + } catch (Exception e) { + LOG.error("Failed to stop websocket-client", e); + } + try { + httpClient.stop(); + } catch (Exception e) { + LOG.error("Failed to stop http-client", e); + } + return returnCode; } - @WebSocket(maxTextMessageSize = 64 * 1024) + @WebSocket public static class ProducerSocket { private final CountDownLatch closeLatch; @@ -538,7 +554,7 @@ public ProducerSocket(CompletableFuture connected) { } public CompletableFuture send(int index, byte[] content) throws Exception { - this.session.getRemote().sendString(getTestJsonPayload(index, content)); + this.session.sendText(getTestJsonPayload(index, content), Callback.NOOP); this.result = new CompletableFuture<>(); return result; } @@ -561,7 +577,7 @@ public void onClose(int statusCode, String reason) { this.closeLatch.countDown(); } - @OnWebSocketConnect + @OnWebSocketOpen public void onConnect(Session session) { LOG.info("Got connect: {}", session); this.session = session; @@ -576,10 +592,6 @@ public synchronized void onMessage(String msg) throws JsonParseException { } } - public RemoteEndpoint getRemote() { - return this.session.getRemote(); - } - public Session getSession() { return this.session; } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdRead.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdRead.java index daab436499219..f75d93cad127c 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdRead.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdRead.java @@ -39,6 +39,7 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.naming.TopicName; +import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; @@ -231,7 +232,9 @@ private int readFromWebSocket(String topic) { URI readerUri = URI.create(getWebSocketReadUri(topic)); - WebSocketClient readClient = new WebSocketClient(new SslContextFactory(true)); + HttpClient httpClient = new HttpClient(); + httpClient.setSslContextFactory(new SslContextFactory.Client(true)); + WebSocketClient readClient = new WebSocketClient(httpClient); ClientUpgradeRequest readRequest = new ClientUpgradeRequest(); try { if (authentication != null) { @@ -293,6 +296,17 @@ private int readFromWebSocket(String topic) { LOG.info("{} messages successfully read", numMessagesRead); } + try { + readClient.stop(); + } catch (Exception e) { + LOG.error("Failed to stop websocket-client", e); + } + try { + httpClient.stop(); + } catch (Exception e) { + LOG.error("Failed to stop http-client", e); + } + return returnCode; } diff --git a/pulsar-functions/utils/pom.xml b/pulsar-functions/utils/pom.xml index 27a771fe652bd..270ce45bf40cf 100644 --- a/pulsar-functions/utils/pom.xml +++ b/pulsar-functions/utils/pom.xml @@ -112,7 +112,7 @@ com.github.tomakehurst - wiremock-jre8 + wiremock-jre8-standalone ${wiremock.version} test diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml index 20ab02f45f809..6d81915b1a2c6 100644 --- a/pulsar-functions/worker/pom.xml +++ b/pulsar-functions/worker/pom.xml @@ -104,13 +104,13 @@ - org.eclipse.jetty - jetty-servlet + org.eclipse.jetty.ee8 + jetty-ee8-servlet - org.eclipse.jetty - jetty-servlets + org.eclipse.jetty.ee8 + jetty-ee8-servlets @@ -162,11 +162,6 @@ - - io.prometheus - simpleclient_jetty - - diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java index 583d8ce558b08..cafb88de219d0 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.functions.worker.rest; -import io.prometheus.client.jetty.JettyStatisticsCollector; import java.util.ArrayList; import java.util.EnumSet; import java.util.List; @@ -34,7 +33,12 @@ import org.apache.pulsar.functions.worker.WorkerService; import org.apache.pulsar.functions.worker.rest.api.v2.WorkerApiV2Resource; import org.apache.pulsar.functions.worker.rest.api.v2.WorkerStatsApiV2Resource; +import org.apache.pulsar.jetty.metrics.JettyStatisticsCollector; import org.apache.pulsar.jetty.tls.JettySslContextFactory; +import org.eclipse.jetty.ee8.servlet.FilterHolder; +import org.eclipse.jetty.ee8.servlet.ServletContextHandler; +import org.eclipse.jetty.ee8.servlet.ServletHolder; +import org.eclipse.jetty.ee8.servlets.QoSFilter; import org.eclipse.jetty.server.ConnectionFactory; import org.eclipse.jetty.server.ConnectionLimit; import org.eclipse.jetty.server.ForwardedRequestCustomizer; @@ -48,13 +52,7 @@ import org.eclipse.jetty.server.SslConnectionFactory; import org.eclipse.jetty.server.handler.ContextHandlerCollection; import org.eclipse.jetty.server.handler.DefaultHandler; -import org.eclipse.jetty.server.handler.HandlerCollection; -import org.eclipse.jetty.server.handler.RequestLogHandler; import org.eclipse.jetty.server.handler.StatisticsHandler; -import org.eclipse.jetty.servlet.FilterHolder; -import org.eclipse.jetty.servlet.ServletContextHandler; -import org.eclipse.jetty.servlet.ServletHolder; -import org.eclipse.jetty.servlets.QoSFilter; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.glassfish.jersey.server.ResourceConfig; import org.glassfish.jersey.servlet.ServletContainer; @@ -116,28 +114,26 @@ private void init() { List handlers = new ArrayList<>(4); handlers.add(newServletContextHandler("/admin", - new ResourceConfig(Resources.getApiV2Resources()), workerService, filterInitializer)); + new ResourceConfig(Resources.getApiV2Resources()), workerService, filterInitializer).get()); handlers.add(newServletContextHandler("/admin/v2", - new ResourceConfig(Resources.getApiV2Resources()), workerService, filterInitializer)); + new ResourceConfig(Resources.getApiV2Resources()), workerService, filterInitializer).get()); handlers.add(newServletContextHandler("/admin/v3", - new ResourceConfig(Resources.getApiV3Resources()), workerService, filterInitializer)); + new ResourceConfig(Resources.getApiV3Resources()), workerService, filterInitializer).get()); // don't require auth for metrics or config routes handlers.add(newServletContextHandler("/", new ResourceConfig(Resources.getRootResources()), workerService, - workerConfig.isAuthenticateMetricsEndpoint(), filterInitializer)); + workerConfig.isAuthenticateMetricsEndpoint(), filterInitializer).get()); - RequestLogHandler requestLogHandler = new RequestLogHandler(); boolean showDetailedAddresses = workerConfig.getWebServiceLogDetailedAddresses() != null ? workerConfig.getWebServiceLogDetailedAddresses() : (workerConfig.isWebServiceHaProxyProtocolEnabled() || workerConfig.isWebServiceTrustXForwardedFor()); - requestLogHandler.setRequestLog(JettyRequestLogFactory.createRequestLogger(showDetailedAddresses, server)); - handlers.add(0, new ContextHandlerCollection()); - handlers.add(requestLogHandler); + server.setRequestLog(JettyRequestLogFactory.createRequestLogger(showDetailedAddresses, server)); + ContextHandlerCollection contexts = new ContextHandlerCollection(); - contexts.setHandlers(handlers.toArray(new Handler[handlers.size()])); - HandlerCollection handlerCollection = new HandlerCollection(); - handlerCollection.setHandlers(new Handler[]{contexts, new DefaultHandler(), requestLogHandler}); + contexts.setHandlers(handlers); + Handler.Collection handlerCollection = new Handler.Sequence(); + handlerCollection.setHandlers(contexts, new DefaultHandler()); // Metrics handler StatisticsHandler stats = new StatisticsHandler(); @@ -147,13 +143,13 @@ private void init() { } catch (IllegalArgumentException e) { // Already registered. Eg: in unit tests } - handlers.add(stats); + server.setHandler(stats); if (this.workerConfig.getTlsEnabled()) { log.info("Configuring https server on port={}", this.workerConfig.getWorkerPortTls()); try { - SslContextFactory sslCtxFactory; + SslContextFactory.Server sslCtxFactory; if (workerConfig.isTlsEnabledWithKeyStore()) { sslCtxFactory = JettySslContextFactory.createServerSslContextWithKeystore( workerConfig.getTlsProvider(), @@ -241,9 +237,9 @@ public void addFilters(ServletContextHandler context, boolean requiresAuthentica } static ServletContextHandler newServletContextHandler(String contextPath, - ResourceConfig config, - WorkerService workerService, - FilterInitializer filterInitializer) { + ResourceConfig config, + WorkerService workerService, + FilterInitializer filterInitializer) { return newServletContextHandler(contextPath, config, workerService, true, filterInitializer); } diff --git a/pulsar-io/alluxio/pom.xml b/pulsar-io/alluxio/pom.xml index 3b7d4e5057d39..8c1d7473a1287 100644 --- a/pulsar-io/alluxio/pom.xml +++ b/pulsar-io/alluxio/pom.xml @@ -33,6 +33,7 @@ 4.1.11 1.37.0 4.1.100.Final + 9.4.54.v20240208 pulsar-io-alluxio @@ -114,6 +115,13 @@ metrics-jvm ${metrics.version} + + org.eclipse.jetty + jetty-bom + ${jetty9.version} + pom + import + diff --git a/pulsar-io/debezium/core/pom.xml b/pulsar-io/debezium/core/pom.xml index d5364b7d36dd4..4aeb691311c5b 100644 --- a/pulsar-io/debezium/core/pom.xml +++ b/pulsar-io/debezium/core/pom.xml @@ -75,6 +75,10 @@ jose4j org.bitbucket.b_c + + org.eclipse.jetty + * + diff --git a/pulsar-io/docs/pom.xml b/pulsar-io/docs/pom.xml index 82c8f0bb6f96a..ab6954161ad97 100644 --- a/pulsar-io/docs/pom.xml +++ b/pulsar-io/docs/pom.xml @@ -127,11 +127,6 @@ pulsar-io-hbase ${project.version} - - ${project.groupId} - pulsar-io-hdfs2 - ${project.version} - ${project.groupId} pulsar-io-hdfs3 diff --git a/pulsar-io/flume/pom.xml b/pulsar-io/flume/pom.xml index 9b2839970ab79..bddc8f8546a9b 100644 --- a/pulsar-io/flume/pom.xml +++ b/pulsar-io/flume/pom.xml @@ -31,6 +31,23 @@ pulsar-io-flume Pulsar IO :: Flume + + 9.4.54.v20240208 + + + + + + + org.eclipse.jetty + jetty-bom + ${jetty9.version} + pom + import + + + + ${project.groupId} diff --git a/pulsar-io/hbase/pom.xml b/pulsar-io/hbase/pom.xml index 0c38d4f06d029..4e7a3fcff3fe0 100644 --- a/pulsar-io/hbase/pom.xml +++ b/pulsar-io/hbase/pom.xml @@ -73,9 +73,13 @@ log4j log4j + + ch.qos.reload4j + reload4j + org.slf4j - slf4j-log4j12 + * diff --git a/pulsar-io/hdfs2/pom.xml b/pulsar-io/hdfs2/pom.xml deleted file mode 100644 index 81b67f8e095fa..0000000000000 --- a/pulsar-io/hdfs2/pom.xml +++ /dev/null @@ -1,131 +0,0 @@ - - - 4.0.0 - - org.apache.pulsar - pulsar-io - 3.4.0-SNAPSHOT - - pulsar-io-hdfs2 - Pulsar IO :: Hdfs2 - - - - ${project.groupId} - pulsar-io-core - ${project.version} - - - - com.fasterxml.jackson.core - jackson-databind - - - - com.fasterxml.jackson.dataformat - jackson-dataformat-yaml - - - - org.apache.commons - commons-collections4 - - - - org.apache.hadoop - hadoop-client - ${hadoop2.version} - - - log4j - log4j - - - org.slf4j - * - - - org.apache.avro - avro - - - - - org.apache.commons - commons-lang3 - - - - - - - org.apache.nifi - nifi-nar-maven-plugin - - - com.github.spotbugs - spotbugs-maven-plugin - ${spotbugs-maven-plugin.version} - - ${basedir}/src/main/resources/findbugsExclude.xml - - - - spotbugs - verify - - check - - - - - - - - - - owasp-dependency-check - - - - org.owasp - dependency-check-maven - ${dependency-check-maven.version} - - - - aggregate - - none - - - - - - - - - \ No newline at end of file diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConfig.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConfig.java deleted file mode 100644 index 757360e04533c..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConfig.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * 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.io.hdfs2; - -import java.io.Serializable; -import lombok.Data; -import lombok.experimental.Accessors; -import org.apache.commons.lang.StringUtils; - -/** - * Configuration object for all HDFS components. - */ -@Data -@Accessors(chain = true) -public abstract class AbstractHdfsConfig implements Serializable { - - private static final long serialVersionUID = 1L; - - /** - * A file or comma separated list of files which contains the Hadoop file system configuration, - * e.g. 'core-site.xml', 'hdfs-site.xml'. - */ - private String hdfsConfigResources; - - /** - * The HDFS directory from which files should be read from or written to. - */ - private String directory; - - /** - * The character encoding for the files, e.g. UTF-8, ASCII, etc. - */ - private String encoding; - - /** - * The compression codec used to compress/de-compress the files on HDFS. - */ - private Compression compression; - - /** - * The Kerberos user principal account to use for authentication. - */ - private String kerberosUserPrincipal; - - /** - * The full pathname to the Kerberos keytab file to use for authentication. - */ - private String keytab; - - public void validate() { - if (StringUtils.isEmpty(hdfsConfigResources) || StringUtils.isEmpty(directory)) { - throw new IllegalArgumentException("Required property not set."); - } - - if ((StringUtils.isNotEmpty(kerberosUserPrincipal) && StringUtils.isEmpty(keytab)) - || (StringUtils.isEmpty(kerberosUserPrincipal) && StringUtils.isNotEmpty(keytab))) { - throw new IllegalArgumentException("Values for both kerberosUserPrincipal & keytab are required."); - } - } -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConnector.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConnector.java deleted file mode 100644 index d7277aa627383..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConnector.java +++ /dev/null @@ -1,246 +0,0 @@ -/* - * 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.io.hdfs2; - -import java.io.IOException; -import java.lang.ref.WeakReference; -import java.net.InetSocketAddress; -import java.net.Socket; -import java.net.URI; -import java.nio.charset.Charset; -import java.security.PrivilegedExceptionAction; -import java.util.Collections; -import java.util.Map; -import java.util.WeakHashMap; -import java.util.concurrent.atomic.AtomicReference; -import javax.net.SocketFactory; -import org.apache.commons.lang.StringUtils; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.compress.CompressionCodec; -import org.apache.hadoop.io.compress.CompressionCodecFactory; -import org.apache.hadoop.io.compress.DefaultCodec; -import org.apache.hadoop.net.NetUtils; -import org.apache.hadoop.security.UserGroupInformation; -import org.apache.pulsar.io.hdfs2.sink.HdfsSinkConfig; - -/** - * A Simple abstract class for HDFS connectors. - * Provides methods for connecting to HDFS - */ -public abstract class AbstractHdfsConnector { - - private static final Object RESOURCES_LOCK = new Object(); - - // Hadoop Configuration, Filesystem, and UserGroupInformation (optional) - protected final AtomicReference hdfsResources = new AtomicReference<>(); - protected AbstractHdfsConfig connectorConfig; - protected CompressionCodecFactory compressionCodecFactory; - - public AbstractHdfsConnector() { - hdfsResources.set(new HdfsResources(null, null, null)); - } - - /* - * Reset Hadoop Configuration and FileSystem based on the supplied configuration resources. - */ - protected HdfsResources resetHDFSResources(HdfsSinkConfig hdfsSinkConfig) throws IOException { - Configuration config = new ExtendedConfiguration(); - config.setClassLoader(Thread.currentThread().getContextClassLoader()); - - getConfig(config, connectorConfig.getHdfsConfigResources()); - - // first check for timeout on HDFS connection, because FileSystem has a hard coded 15 minute timeout - checkHdfsUriForTimeout(config); - - /* Disable caching of Configuration and FileSystem objects, else we cannot reconfigure - * the processor without a complete restart - */ - String disableCacheName = String.format("fs.%s.impl.disable.cache", - FileSystem.getDefaultUri(config).getScheme()); - config.set(disableCacheName, "true"); - - // If kerberos is enabled, create the file system as the kerberos principal - // -- use RESOURCE_LOCK to guarantee UserGroupInformation is accessed by only a single thread at at time - FileSystem fs; - UserGroupInformation ugi; - synchronized (RESOURCES_LOCK) { - if (SecurityUtil.isSecurityEnabled(config)) { - ugi = SecurityUtil.loginKerberos(config, - connectorConfig.getKerberosUserPrincipal(), connectorConfig.getKeytab()); - fs = getFileSystemAsUser(config, ugi); - } else { - config.set("ipc.client.fallback-to-simple-auth-allowed", "true"); - config.set("hadoop.security.authentication", "simple"); - ugi = SecurityUtil.loginSimple(config); - fs = getFileSystemAsUser(config, ugi); - } - } - return new HdfsResources(config, fs, ugi); - } - - private static Configuration getConfig(final Configuration config, String res) throws IOException { - boolean foundResources = false; - if (null != res) { - String[] resources = res.split(","); - for (String resource : resources) { - config.addResource(new Path(resource.trim())); - foundResources = true; - } - } - - if (!foundResources) { - // check that at least 1 non-default resource is available on the classpath - String configStr = config.toString(); - for (String resource : configStr.substring(configStr.indexOf(":") + 1).split(",")) { - if (!resource.contains("default") && config.getResource(resource.trim()) != null) { - foundResources = true; - break; - } - } - } - - if (!foundResources) { - throw new IOException("Could not find any of the " + res + " on the classpath"); - } - return config; - } - - /* - * Reduce the timeout of a socket connection from the default in FileSystem.get() - */ - protected void checkHdfsUriForTimeout(Configuration config) throws IOException { - URI hdfsUri = FileSystem.getDefaultUri(config); - String address = hdfsUri.getAuthority(); - int port = hdfsUri.getPort(); - if (address == null || address.isEmpty() || port < 0) { - return; - } - InetSocketAddress namenode = NetUtils.createSocketAddr(address, port); - SocketFactory socketFactory = NetUtils.getDefaultSocketFactory(config); - try (Socket socket = socketFactory.createSocket()) { - NetUtils.connect(socket, namenode, 1000); // 1 second timeout - } - } - - /** - * This exists in order to allow unit tests to override it so that they don't take several - * minutes waiting for UDP packets to be received. - * - * @param config - * the configuration to use - * @return the FileSystem that is created for the given Configuration - * @throws IOException - * if unable to create the FileSystem - */ - protected FileSystem getFileSystem(final Configuration config) throws IOException { - return FileSystem.get(config); - } - - protected FileSystem getFileSystemAsUser(final Configuration config, UserGroupInformation ugi) throws IOException { - try { - return ugi.doAs((PrivilegedExceptionAction) () -> FileSystem.get(config)); - } catch (InterruptedException e) { - throw new IOException("Unable to create file system: " + e.getMessage()); - } - } - - protected Configuration getConfiguration() { - return hdfsResources.get().getConfiguration(); - } - - protected FileSystem getFileSystem() { - return hdfsResources.get().getFileSystem(); - } - - protected UserGroupInformation getUserGroupInformation() { - return hdfsResources.get().getUserGroupInformation(); - } - - protected String getEncoding() { - return StringUtils.isNotBlank(connectorConfig.getEncoding()) - ? connectorConfig.getEncoding() : Charset.defaultCharset().name(); - } - - protected CompressionCodec getCompressionCodec() { - if (connectorConfig.getCompression() == null) { - return null; - } - - CompressionCodec codec = getCompressionCodecFactory() - .getCodecByName(connectorConfig.getCompression().name()); - - return (codec != null) ? codec : new DefaultCodec(); - } - - protected CompressionCodecFactory getCompressionCodecFactory() { - if (compressionCodecFactory == null) { - compressionCodecFactory = new CompressionCodecFactory(getConfiguration()); - } - - return compressionCodecFactory; - } - - /** - * Extending Hadoop Configuration to prevent it from caching classes that can't be found. Since users may be - * adding additional JARs to the classpath we don't want them to have to restart the JVM to be able to load - * something that was previously not found, but might now be available. - * Reference the original getClassByNameOrNull from Configuration. - */ - static class ExtendedConfiguration extends Configuration { - - private final Map>>> cacheClasses = new WeakHashMap<>(); - - @Override - public Class getClassByNameOrNull(String name) { - final ClassLoader classLoader = getClassLoader(); - - Map>> map; - synchronized (cacheClasses) { - map = cacheClasses.get(classLoader); - if (map == null) { - map = Collections.synchronizedMap(new WeakHashMap<>()); - cacheClasses.put(classLoader, map); - } - } - - Class clazz = null; - WeakReference> ref = map.get(name); - if (ref != null) { - clazz = ref.get(); - } - - if (clazz == null) { - try { - clazz = Class.forName(name, true, classLoader); - } catch (ClassNotFoundException | NoClassDefFoundError e) { - return null; - } - // two putters can race here, but they'll put the same class - map.put(name, new WeakReference<>(clazz)); - return clazz; - } else { - // cache hit - return clazz; - } - } - - } -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/Compression.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/Compression.java deleted file mode 100644 index 1e3d2f9490439..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/Compression.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * 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.io.hdfs2; - -/** - * An enumeration of compression codecs available for HDFS. - */ -public enum Compression { - BZIP2, DEFLATE, GZIP, LZ4, SNAPPY, ZSTANDARD -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/HdfsResources.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/HdfsResources.java deleted file mode 100644 index 5fd6b283e6b41..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/HdfsResources.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * 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.io.hdfs2; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.security.UserGroupInformation; - -/** - * A wrapper class for HDFS resources. - */ -public class HdfsResources { - - private final Configuration configuration; - private final FileSystem fileSystem; - private final UserGroupInformation userGroupInformation; - - public HdfsResources(Configuration config, FileSystem fs, UserGroupInformation ugi) { - this.configuration = config; - this.fileSystem = fs; - this.userGroupInformation = ugi; - } - - public Configuration getConfiguration() { - return configuration; - } - - public FileSystem getFileSystem() { - return fileSystem; - } - - public UserGroupInformation getUserGroupInformation() { - return userGroupInformation; - } -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/SecurityUtil.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/SecurityUtil.java deleted file mode 100644 index ca178aad911e2..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/SecurityUtil.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * 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.io.hdfs2; - -import java.io.IOException; -import org.apache.commons.lang3.Validate; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.security.UserGroupInformation; - -/** - * Provides synchronized access to UserGroupInformation to avoid multiple processors/services from - * interfering with each other. - */ -public class SecurityUtil { - public static final String HADOOP_SECURITY_AUTHENTICATION = "hadoop.security.authentication"; - public static final String KERBEROS = "kerberos"; - - /** - * Initializes UserGroupInformation with the given Configuration and performs the login for the - * given principal and keytab. All logins should happen through this class to ensure other threads - * are not concurrently modifying UserGroupInformation. - *

- * @param config the configuration instance - * @param principal the principal to authenticate as - * @param keyTab the keytab to authenticate with - * - * @return the UGI for the given principal - * - * @throws IOException if login failed - */ - public static synchronized UserGroupInformation loginKerberos(final Configuration config, - final String principal, final String keyTab) throws IOException { - Validate.notNull(config); - Validate.notNull(principal); - Validate.notNull(keyTab); - - UserGroupInformation.setConfiguration(config); - UserGroupInformation.loginUserFromKeytab(principal.trim(), keyTab.trim()); - return UserGroupInformation.getCurrentUser(); - } - - /** - * Initializes UserGroupInformation with the given Configuration and - * returns UserGroupInformation.getLoginUser(). All logins should happen - * through this class to ensure other threads are not concurrently - * modifying UserGroupInformation. - * - * @param config the configuration instance - * - * @return the UGI for the given principal - * - * @throws IOException if login failed - */ - public static synchronized UserGroupInformation loginSimple(final Configuration config) throws IOException { - Validate.notNull(config); - UserGroupInformation.setConfiguration(config); - return UserGroupInformation.getLoginUser(); - } - - /** - * Initializes UserGroupInformation with the given Configuration and returns - * UserGroupInformation.isSecurityEnabled(). - * All checks for isSecurityEnabled() should happen through this method. - * - * @param config the given configuration - * - * @return true if kerberos is enabled on the given configuration, false otherwise - * - */ - public static boolean isSecurityEnabled(final Configuration config) { - Validate.notNull(config); - return KERBEROS.equalsIgnoreCase(config.get(HADOOP_SECURITY_AUTHENTICATION)); - } -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsAbstractSink.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsAbstractSink.java deleted file mode 100644 index 7b025d16378ff..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsAbstractSink.java +++ /dev/null @@ -1,124 +0,0 @@ -/* - * 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.io.hdfs2.sink; - -import java.io.IOException; -import java.time.LocalDateTime; -import java.time.format.DateTimeFormatter; -import java.util.Map; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.io.FilenameUtils; -import org.apache.commons.lang.StringUtils; -import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.pulsar.functions.api.Record; -import org.apache.pulsar.io.core.KeyValue; -import org.apache.pulsar.io.core.Sink; -import org.apache.pulsar.io.core.SinkContext; -import org.apache.pulsar.io.hdfs2.AbstractHdfsConnector; -import org.apache.pulsar.io.hdfs2.HdfsResources; - -/** - * A Simple abstract class for HDFS sink. - * Users need to implement extractKeyValue function to use this sink. - */ -@Slf4j -public abstract class HdfsAbstractSink extends AbstractHdfsConnector implements Sink { - - protected HdfsSinkConfig hdfsSinkConfig; - protected BlockingQueue> unackedRecords; - protected HdfsSyncThread syncThread; - private Path path; - private FSDataOutputStream hdfsStream; - private DateTimeFormatter subdirectoryFormatter; - - public abstract KeyValue extractKeyValue(Record record); - protected abstract void createWriter() throws IOException; - - @Override - public void open(Map config, SinkContext sinkContext) throws Exception { - hdfsSinkConfig = HdfsSinkConfig.load(config); - hdfsSinkConfig.validate(); - connectorConfig = hdfsSinkConfig; - unackedRecords = new LinkedBlockingQueue> (hdfsSinkConfig.getMaxPendingRecords()); - if (hdfsSinkConfig.getSubdirectoryPattern() != null) { - subdirectoryFormatter = DateTimeFormatter.ofPattern(hdfsSinkConfig.getSubdirectoryPattern()); - } - connectToHdfs(); - createWriter(); - launchSyncThread(); - } - - @Override - public void close() throws Exception { - syncThread.halt(); - syncThread.join(0); - } - - protected final void connectToHdfs() throws IOException { - try { - HdfsResources resources = hdfsResources.get(); - - if (resources.getConfiguration() == null) { - resources = this.resetHDFSResources(hdfsSinkConfig); - hdfsResources.set(resources); - } - } catch (IOException ex) { - hdfsResources.set(new HdfsResources(null, null, null)); - throw ex; - } - } - - protected FSDataOutputStream getHdfsStream() throws IllegalArgumentException, IOException { - if (hdfsStream == null) { - Path path = getPath(); - FileSystem fs = getFileSystemAsUser(getConfiguration(), getUserGroupInformation()); - hdfsStream = fs.exists(path) ? fs.append(path) : fs.create(path); - } - return hdfsStream; - } - - protected final Path getPath() { - if (path == null) { - String ext = ""; - if (StringUtils.isNotBlank(hdfsSinkConfig.getFileExtension())) { - ext = hdfsSinkConfig.getFileExtension(); - } else if (getCompressionCodec() != null) { - ext = getCompressionCodec().getDefaultExtension(); - } - - String directory = hdfsSinkConfig.getDirectory(); - if (subdirectoryFormatter != null) { - directory = FilenameUtils.concat(directory, LocalDateTime.now().format(subdirectoryFormatter)); - } - path = new Path(FilenameUtils.concat(directory, - hdfsSinkConfig.getFilenamePrefix() + "-" + System.currentTimeMillis() + ext)); - log.info("Create path: {}", path); - } - return path; - } - - protected final void launchSyncThread() throws IOException { - syncThread = new HdfsSyncThread(getHdfsStream(), unackedRecords, hdfsSinkConfig.getSyncInterval()); - syncThread.start(); - } -} \ No newline at end of file diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsSinkConfig.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsSinkConfig.java deleted file mode 100644 index 9e1c6090fb5b7..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsSinkConfig.java +++ /dev/null @@ -1,117 +0,0 @@ -/* - * 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.io.hdfs2.sink; - -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; -import java.io.File; -import java.io.IOException; -import java.io.Serializable; -import java.time.LocalDateTime; -import java.time.format.DateTimeFormatter; -import java.util.Map; -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.experimental.Accessors; -import org.apache.commons.lang.StringUtils; -import org.apache.pulsar.io.hdfs2.AbstractHdfsConfig; - -/** - * Configuration object for all HDFS Sink components. - */ -@Data -@EqualsAndHashCode(callSuper = false) -@Accessors(chain = true) -public class HdfsSinkConfig extends AbstractHdfsConfig implements Serializable { - - private static final long serialVersionUID = 1L; - - /** - * The prefix of the files to create inside the HDFS directory, i.e. a value of "topicA" - * will result in files named topicA-, topicA-, etc being produced - */ - private String filenamePrefix; - - /** - * The extension to add to the files written to HDFS, e.g. '.txt', '.seq', etc. - */ - private String fileExtension; - - /** - * The character to use to separate records in a text file. If no value is provided - * then the content from all of the records will be concatenated together in one continuous - * byte array. - */ - private char separator; - - /** - * The interval (in milliseconds) between calls to flush data to HDFS disk. - */ - private long syncInterval; - - /** - * The maximum number of records that we hold in memory before acking. Default is Integer.MAX_VALUE. - * Setting this value to one, results in every record being sent to disk before the record is acked, - * while setting it to a higher values allows us to buffer records before flushing them all to disk. - */ - private int maxPendingRecords = Integer.MAX_VALUE; - - /** - * A subdirectory associated with the created time of the sink. - * The pattern is the formatted pattern of {@link AbstractHdfsConfig#getDirectory()}'s subdirectory. - * - * @see java.time.format.DateTimeFormatter for pattern's syntax - */ - private String subdirectoryPattern; - - public static HdfsSinkConfig load(String yamlFile) throws IOException { - ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); - return mapper.readValue(new File(yamlFile), HdfsSinkConfig.class); - } - - public static HdfsSinkConfig load(Map map) throws IOException { - ObjectMapper mapper = new ObjectMapper(); - return mapper.readValue(mapper.writeValueAsString(map), HdfsSinkConfig.class); - } - - @Override - public void validate() { - super.validate(); - if ((StringUtils.isEmpty(fileExtension) && getCompression() == null) - || StringUtils.isEmpty(filenamePrefix)) { - throw new IllegalArgumentException("Required property not set."); - } - - if (syncInterval < 0) { - throw new IllegalArgumentException("Sync Interval cannot be negative"); - } - - if (maxPendingRecords < 1) { - throw new IllegalArgumentException("Max Pending Records must be a positive integer"); - } - - if (subdirectoryPattern != null) { - try { - LocalDateTime.of(2020, 1, 1, 12, 0).format(DateTimeFormatter.ofPattern(subdirectoryPattern)); - } catch (Exception e) { - throw new IllegalArgumentException(subdirectoryPattern + " is not a valid pattern: " + e.getMessage()); - } - } - } -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsSyncThread.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsSyncThread.java deleted file mode 100644 index 9ddd83f4423f9..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/HdfsSyncThread.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * 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.io.hdfs2.sink; - -import java.io.IOException; -import java.util.concurrent.BlockingQueue; -import org.apache.commons.collections4.CollectionUtils; -import org.apache.hadoop.fs.Syncable; -import org.apache.pulsar.functions.api.Record; - -/** - * A thread that runs in the background and acknowledges Records - * after they have been written to disk. - * - * @param - */ -public class HdfsSyncThread extends Thread { - - private final Syncable stream; - private final BlockingQueue> unackedRecords; - private final long syncInterval; - private boolean keepRunning = true; - - public HdfsSyncThread(Syncable stream, BlockingQueue> unackedRecords, long syncInterval) { - this.stream = stream; - this.unackedRecords = unackedRecords; - this.syncInterval = syncInterval; - } - - @Override - public void run() { - while (keepRunning) { - try { - Thread.sleep(syncInterval); - ackRecords(); - } catch (InterruptedException e) { - return; - } catch (IOException e) { - e.printStackTrace(); - } - } - } - - public final void halt() throws IOException, InterruptedException { - keepRunning = false; - ackRecords(); - } - - private void ackRecords() throws IOException, InterruptedException { - - if (CollectionUtils.isEmpty(unackedRecords)) { - return; - } - - synchronized (stream) { - stream.hsync(); - } - - while (!unackedRecords.isEmpty()) { - unackedRecords.take().ack(); - } - } -} diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/package-info.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/package-info.java deleted file mode 100644 index 238a441ee0ee3..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/package-info.java +++ /dev/null @@ -1,19 +0,0 @@ -/* - * 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.io.hdfs2.sink; \ No newline at end of file diff --git a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/seq/HdfsAbstractSequenceFileSink.java b/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/seq/HdfsAbstractSequenceFileSink.java deleted file mode 100644 index 355c00080effe..0000000000000 --- a/pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/sink/seq/HdfsAbstractSequenceFileSink.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * 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.io.hdfs2.sink.seq; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.atomic.AtomicLong; -import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.io.SequenceFile; -import org.apache.hadoop.io.SequenceFile.Writer; -import org.apache.hadoop.io.SequenceFile.Writer.Option; -import org.apache.pulsar.functions.api.Record; -import org.apache.pulsar.io.core.KeyValue; -import org.apache.pulsar.io.core.Sink; -import org.apache.pulsar.io.hdfs2.sink.HdfsAbstractSink; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * HDFS Sink that writes it contents to HDFS as Sequence Files. - * - * @param - The incoming Key type - * @param - The incoming Value type - * @param - The HDFS Key type - * @param - The HDFS Value type - */ -public abstract class HdfsAbstractSequenceFileSink - extends HdfsAbstractSink implements Sink { - - private static final Logger LOG = LoggerFactory.getLogger(HdfsAbstractSequenceFileSink.class); - - protected AtomicLong counter; - protected FSDataOutputStream hdfsStream; - protected Writer writer = null; - - public abstract KeyValue convert(KeyValue kv); - - @Override - public void close() throws Exception { - writer.close(); - super.close(); - } - - @Override - protected void createWriter() throws IOException { - writer = getWriter(); - } - - @Override - public void write(Record record) { - try { - KeyValue kv = extractKeyValue(record); - KeyValue keyValue = convert(kv); - writer.append(keyValue.getKey(), keyValue.getValue()); - unackedRecords.put(record); - } catch (IOException | InterruptedException e) { - LOG.error("Unable to write to file " + getPath(), e); - record.fail(); - } - } - - protected Writer getWriter() throws IOException { - counter = new AtomicLong(0); - List