Revert "Update dependencies to resolve vulnerabilities (#183)"

This reverts commit 0bb284e.

Author: nlu90

pom.xml

@@ -66,22 +66,21 @@
 
     <!-- dependencies -->
     <!-- latest version from apache pulsar -->
-    <pulsar.version>4.0.5</pulsar.version>
+    <pulsar.version>2.10.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.19.0</commons-io.version>
+    <commons-io.version>2.11.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.6.0</maven-shade-plugin.version>
+    <maven-shade-plugin.version>3.4.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>
@@ -155,12 +154,6 @@
       <version>${commons-io.version}</version>
     </dependency>
 
-    <dependency>
-      <groupId>org.bouncycastle</groupId>
-      <artifactId>bcprov-jdk18on</artifactId>
-      <version>${bouncycastle.version}</version>
-    </dependency>
-
     <!-- spark dependency -->
 
     <dependency>
@@ -396,7 +389,6 @@
                   <include>org.bouncycastle*:*</include>
                   <include>org.lz4*:*</include>
                   <include>commons-io:commons-io:jar:*</include>
-                  <include>io.opentelemetry:*</include> <!-- Add this -->
                 </includes>
               </artifactSet>
               <filters>
@@ -416,10 +408,6 @@
                 </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,10 +311,8 @@ 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(
-        // passing an empty topicsHash because we don't cache the GetTopicsResponse
-        dest.getNamespaceObject, CommandGetTopicsOfNamespace.Mode.ALL, topicsPattern, "")
-      .get().getTopics
+      .getTopicsUnderNamespace(dest.getNamespaceObject, CommandGetTopicsOfNamespace.Mode.ALL)
+      .get()
 
     val allNonPartitionedTopics: ju.List[String] = allTopics.asScala
       .filter(t => !TopicName.get(t).isPartitioned)
@@ -350,9 +348,7 @@ private[pulsar] case class PulsarHelper(
       while (waitList.nonEmpty) {
         val topic = waitList.head
         try {
-          // setting metadataAutoCreationEnabled to false, and useFallbackForNonPIP344Brokers
-          // to true to conform to non-breaking behavior.
-          client.getPartitionedTopicMetadata(topic, false, true).get()
+          client.getPartitionedTopicMetadata(topic).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,9 +48,10 @@ class PulsarAdmissionControlSuite extends PulsarSourceTest {
     // Need to call latestOffsetForTopicPartition so the helper instantiates
     // the admin
     val admissionControlHelper = new PulsarAdmissionControlHelper(adminUrl, conf)
-    val e = intercept[NotFoundException] {
+    val e = intercept[RuntimeException] {
       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 = "4.0.5"
+  val CURRENT_VERSION = "2.10.2"
 
   var pulsarContainer: PulsarContainer = null
   var serviceUrl: String = null