[Improve] remove admin parameter from sinks and admin options from config

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

@@ -21,8 +21,6 @@ import org.apache.pulsar.common.naming.TopicName
 private[pulsar] object PulsarOptions {
 
   // option key prefix for different modules
-
-  val PulsarAdminOptionKeyPrefix: String = "pulsar.admin."
   val PulsarClientOptionKeyPrefix: String = "pulsar.client."
   val PulsarProducerOptionKeyPrefix: String = "pulsar.producer."
   val PulsarReaderOptionKeyPrefix: String = "pulsar.reader."
@@ -38,7 +36,6 @@ private[pulsar] object PulsarOptions {
   val TopicOptionKeys: Set[String] = Set(TopicSingle, TopicMulti, TopicPattern)
 
   val ServiceUrlOptionKey: String = "service.url"
-  val AdminUrlOptionKey: String = "admin.url"
   val StartingOffsetsOptionKey: String = "startingOffsets".toLowerCase(Locale.ROOT)
   val StartingTime: String = "startingTime".toLowerCase(Locale.ROOT)
   val EndingTime: String = "endingTime".toLowerCase(Locale.ROOT)

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

@@ -185,16 +185,15 @@ private[pulsar] class PulsarProvider
 
     val caseInsensitiveParams = validateSinkOptions(parameters)
 
-    val (clientConfig, producerConfig, topic, adminUrl) = prepareConfForProducer(parameters)
+    val (clientConfig, producerConfig, topic) = prepareConfForProducer(parameters)
     PulsarSinks.validateQuery(data.schema.toAttributes, topic)
 
     PulsarSinks.write(
       sqlContext.sparkSession,
       data.queryExecution,
       clientConfig,
       producerConfig,
-      topic,
-      adminUrl)
+      topic)
 
     /**
      * This method is suppose to return a relation the data that was written. Currently we haven't
@@ -227,9 +226,9 @@ private[pulsar] class PulsarProvider
 
     val caseInsensitiveParams = validateSinkOptions(parameters)
 
-    val (clientConfig, producerConfig, topic, adminUrl) = prepareConfForProducer(parameters)
+    val (clientConfig, producerConfig, topic) = prepareConfForProducer(parameters)
 
-    new PulsarSink(sqlContext, clientConfig, producerConfig, topic, adminUrl)
+    new PulsarSink(sqlContext, clientConfig, producerConfig, topic)
   }
 }
 
@@ -370,10 +369,6 @@ private[pulsar] object PulsarProvider extends Logging {
     parameters(ServiceUrlOptionKey)
   }
 
-  private def getAdminUrl(parameters: Map[String, String]): String = {
-    parameters(AdminUrlOptionKey)
-  }
-
   private def getAllowDifferentTopicSchemas(parameters: Map[String, String]): Boolean = {
     parameters.getOrElse(AllowDifferentTopicSchemas, "false").toBoolean
   }
@@ -481,10 +476,6 @@ private[pulsar] object PulsarProvider extends Logging {
       throw new IllegalArgumentException(s"$ServiceUrlOptionKey must be specified")
     }
 
-    if (!caseInsensitiveParams.contains(AdminUrlOptionKey)) {
-      throw new IllegalArgumentException(s"$AdminUrlOptionKey must be specified")
-    }
-
     val topicOptions =
       caseInsensitiveParams.filter { case (k, _) => TopicOptionKeys.contains(k) }.toSeq.toMap
     if (topicOptions.size > 1 || topicOptions.contains(TopicMulti) || topicOptions.contains(
@@ -512,10 +503,9 @@ private[pulsar] object PulsarProvider extends Logging {
   }
 
   private def prepareConfForProducer(parameters: Map[String, String])
-      : (ju.Map[String, Object], ju.Map[String, Object], Option[String], String) = {
+      : (ju.Map[String, Object], ju.Map[String, Object], Option[String]) = {
 
     val serviceUrl = getServiceUrl(parameters)
-    val adminUrl = getAdminUrl(parameters)
 
     var clientParams = getClientParams(parameters)
     clientParams += (ServiceUrlOptionKey -> serviceUrl)
@@ -526,8 +516,7 @@ private[pulsar] object PulsarProvider extends Logging {
     (
       paramsToPulsarConf("pulsar.client", clientParams),
       paramsToPulsarConf("pulsar.producer", producerParams),
-      topic,
-      adminUrl)
+      topic)
   }
 
   private def jsonOptions: JSONOptionsInRead = {

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

@@ -33,8 +33,7 @@ private[pulsar] class PulsarSink(
     sqlContext: SQLContext,
     pulsarClientConf: ju.Map[String, Object],
     pulsarProducerConf: ju.Map[String, Object],
-    topic: Option[String],
-    adminUrl: String)
+    topic: Option[String])
     extends Sink
     with Logging {
 
@@ -53,8 +52,7 @@ private[pulsar] class PulsarSink(
         data.queryExecution,
         pulsarClientConf,
         pulsarProducerConf,
-        topic,
-        adminUrl)
+        topic)
       latestBatchId = batchId
     }
   }
@@ -141,8 +139,7 @@ private[pulsar] object PulsarSinks extends Logging {
       queryExecution: QueryExecution,
       pulsarClientConf: ju.Map[String, Object],
       pulsarProducerConf: ju.Map[String, Object],
-      topic: Option[String],
-      adminUrl: String): Unit = {
+      topic: Option[String]): Unit = {
 
     // validate the schema
     val schema = queryExecution.analyzed.output
@@ -151,7 +148,7 @@ private[pulsar] object PulsarSinks extends Logging {
     // execute RDD
     queryExecution.toRdd.foreachPartition { iter =>
       val writeTask =
-        new PulsarWriteTask(pulsarClientConf, pulsarProducerConf, topic, schema, adminUrl)
+        new PulsarWriteTask(pulsarClientConf, pulsarProducerConf, topic, schema)
       Utils.tryWithSafeFinally(block = writeTask.execute(iter))(finallyBlock = writeTask.close())
     }
   }

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

@@ -27,9 +27,8 @@ private[pulsar] class PulsarWriteTask(
     clientConf: ju.Map[String, Object],
     producerConf: ju.Map[String, Object],
     topic: Option[String],
-    inputSchema: Seq[Attribute],
-    adminUrl: String)
-    extends PulsarRowWriter(inputSchema, clientConf, producerConf, topic, adminUrl) {
+    inputSchema: Seq[Attribute])
+    extends PulsarRowWriter(inputSchema, clientConf, producerConf, topic) {
 
   /**
    * Writes key value data out to topics.
@@ -53,13 +52,10 @@ private[pulsar] abstract class PulsarRowWriter(
     inputSchema: Seq[Attribute],
     clientConf: ju.Map[String, Object],
     producerConf: ju.Map[String, Object],
-    topic: Option[String],
-    adminUrl: String) {
+    topic: Option[String]) {
 
   import PulsarOptions._
 
-  protected lazy val admin = AdminUtils.buildAdmin(adminUrl, clientConf)
-
   private def createProjections = {
     val topicExpression = topic
       .map(Literal(_))
@@ -213,6 +209,5 @@ private[pulsar] abstract class PulsarRowWriter(
   protected def producerClose(): Unit = {
     producerFlush()
     topic2Producer.clear()
-    admin.close()
   }
 }