Update dependencies to resolve vulnerabilities (#183)

* update pulsar-client

* pulsar-client 2nd change

* upgrade direct dependencies

* bouncycastle: resolve build error

* upgrade pulsar to 4.0.5

* fix PulsarTest

---------

Co-authored-by: Zikang Han <zikang.han@your.hostname.com>

Author: 2 people

pom.xml

@@ -66,21 +66,22 @@
 
     <!-- dependencies -->
     <!-- latest version from apache pulsar -->
-    <pulsar.version>2.10.5</pulsar.version>
+    <pulsar.version>4.0.5</pulsar.version>
     <scala.version>2.13.12</scala.version>
     <scala.binary.version>2.13</scala.binary.version>
     <scalatest.version>3.2.14</scalatest.version>
     <spark.version>3.4.1</spark.version>
-    <commons-io.version>2.11.0</commons-io.version>
+    <commons-io.version>2.19.0</commons-io.version>
     <testcontainers.version>1.18.3</testcontainers.version>
+    <bouncycastle.version>1.78</bouncycastle.version>
 
     <!-- plugin dependencies -->
     <maven.version>3.5.4</maven.version>
     <mvn-scalafmt.version>1.1.1640084764.9f463a9</mvn-scalafmt.version>
     <license-maven-plugin.version>4.1</license-maven-plugin.version>
     <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
     <maven-dependency-plugin.version>3.3.0</maven-dependency-plugin.version>
-    <maven-shade-plugin.version>3.4.0</maven-shade-plugin.version>
+    <maven-shade-plugin.version>3.6.0</maven-shade-plugin.version>
     <maven-source-plugin.version>3.2.1</maven-source-plugin.version>
     <maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
     <scala-maven-plugin.version>4.8.1</scala-maven-plugin.version>
@@ -154,6 +155,12 @@
       <version>${commons-io.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcprov-jdk18on</artifactId>
+      <version>${bouncycastle.version}</version>
+    </dependency>
+
     <!-- spark dependency -->
 
     <dependency>
@@ -389,6 +396,7 @@
                   <include>org.bouncycastle*:*</include>
                   <include>org.lz4*:*</include>
                   <include>commons-io:commons-io:jar:*</include>
+                  <include>io.opentelemetry:*</include> <!-- Add this -->
                 </includes>
               </artifactSet>
               <filters>
@@ -408,6 +416,10 @@
                 </filter>
               </filters>
               <relocations>
+                <relocation>
+                  <pattern>io.opentelemetry</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.io.opentelemetry</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>com.google</pattern>
                   <shadedPattern>org.apache.pulsar.shade.com.google</shadedPattern>

src/main/scala/org/apache/spark/sql/pulsar/PulsarConfigurationUtils.scala

@@ -18,12 +18,12 @@ import java.util.Locale
 
 import scala.reflect._
 
+import com.fasterxml.jackson.annotation.JsonIgnore
 import org.apache.pulsar.client.impl.conf.{
   ClientConfigurationData,
   ProducerConfigurationData,
   ReaderConfigurationData
 }
-import org.apache.pulsar.shade.com.fasterxml.jackson.annotation.JsonIgnore
 
 object PulsarConfigurationUtils {
 

src/main/scala/org/apache/spark/sql/pulsar/PulsarHelper.scala

@@ -311,8 +311,10 @@ private[pulsar] case class PulsarHelper(
   private def getTopics(topicsPattern: String): Seq[String] = {
     val dest = TopicName.get(topicsPattern)
     val allTopics: ju.List[String] = client.getLookup
-      .getTopicsUnderNamespace(dest.getNamespaceObject, CommandGetTopicsOfNamespace.Mode.ALL)
-      .get()
+      .getTopicsUnderNamespace(
+        // passing an empty topicsHash because we don't cache the GetTopicsResponse
+        dest.getNamespaceObject, CommandGetTopicsOfNamespace.Mode.ALL, topicsPattern, "")
+      .get().getTopics
 
     val allNonPartitionedTopics: ju.List[String] = allTopics.asScala
       .filter(t => !TopicName.get(t).isPartitioned)
@@ -348,7 +350,9 @@ private[pulsar] case class PulsarHelper(
       while (waitList.nonEmpty) {
         val topic = waitList.head
         try {
-          client.getPartitionedTopicMetadata(topic).get()
+          // setting metadataAutoCreationEnabled to false, and useFallbackForNonPIP344Brokers
+          // to true to conform to non-breaking behavior.
+          client.getPartitionedTopicMetadata(topic, false, true).get()
           waitList -= topic
         } catch {
           case NonFatal(_) =>

src/test/scala/org/apache/spark/sql/pulsar/PulsarAdmissionControlSuite.scala

@@ -4,10 +4,10 @@ import java.{util => ju}
 
 import org.apache.pulsar.client.admin.PulsarAdmin
 import org.apache.pulsar.client.api.MessageId
-import org.apache.pulsar.client.internal.DefaultImplementation
 import org.apache.spark.sql.pulsar.PulsarSourceUtils.{getEntryId, getLedgerId}
 import org.apache.spark.sql.streaming.Trigger.{Once, ProcessingTime}
 import org.apache.spark.util.Utils
+import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException
 
 class PulsarAdmissionControlSuite extends PulsarSourceTest {
 
@@ -48,10 +48,9 @@ class PulsarAdmissionControlSuite extends PulsarSourceTest {
     // Need to call latestOffsetForTopicPartition so the helper instantiates
     // the admin
     val admissionControlHelper = new PulsarAdmissionControlHelper(adminUrl, conf)
-    val e = intercept[RuntimeException] {
+    val e = intercept[NotFoundException] {
       admissionControlHelper.latestOffsetForTopicPartition(topic, MessageId.earliest, approxSizeOfInt)
     }
-    assert(e.getMessage.contains("Failed to load config into existing configuration data"))
   }
 
   test("Admit entry in the middle of the ledger") {

src/test/scala/org/apache/spark/sql/pulsar/PulsarTest.scala

@@ -49,7 +49,7 @@ trait PulsarTest extends BeforeAndAfterAll with BeforeAndAfterEach {
   self: SparkFunSuite =>
   import PulsarOptions._
 
-  val CURRENT_VERSION = "2.10.2"
+  val CURRENT_VERSION = "4.0.5"
 
   var pulsarContainer: PulsarContainer = null
   var serviceUrl: String = null