From 8ecf6e12312d380cb2c02e57ef432f9d45fa09dd Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Thu, 14 May 2026 15:16:34 +0800
Subject: [PATCH 1/7] Add Pulsar AutoClusterFailover support

---
 README.md                                     |  38 +++
 .../spark/sql/pulsar/CachedPulsarClient.scala |  31 +-
 .../sql/pulsar/PulsarFailoverConfig.scala     | 308 ++++++++++++++++++
 .../spark/sql/pulsar/PulsarOptions.scala      |  13 +
 .../spark/sql/pulsar/PulsarProvider.scala     |  43 ++-
 .../pulsar/PulsarFailoverConfigSuite.scala    | 132 ++++++++
 6 files changed, 552 insertions(+), 13 deletions(-)
 create mode 100644 src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
 create mode 100644 src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala

diff --git a/README.md b/README.md
index 3937d5f..af16464 100644
--- a/README.md
+++ b/README.md
@@ -502,6 +502,44 @@ Please check [Pulsar Producer Configuration](https://pulsar.apache.org/docs/2.11
 
 </table>
 
+
+### AutoClusterFailover
+
+The connector can configure Pulsar client-side `AutoClusterFailover` through DataFrame options. Set `pulsar.failover.primary.serviceUrl` to enable failover. When failover is enabled, `service.url` is optional; if both are set, they must match.
+
+```scala
+val df = spark.readStream
+  .format("pulsar")
+  .option("pulsar.failover.primary.serviceUrl", "pulsar://primary:6650")
+  .option("pulsar.failover.secondary.0.serviceUrl", "pulsar://secondary-a:6650")
+  .option("pulsar.failover.secondary.1.serviceUrl", "pulsar://secondary-b:6650")
+  .option("pulsar.failover.failoverDelayMs", "30000")
+  .option("pulsar.failover.switchBackDelayMs", "60000")
+  .option("pulsar.failover.checkIntervalMs", "30000")
+  .option("topic", "persistent://public/default/input")
+  .load()
+```
+
+Primary cluster authentication and TLS use existing `pulsar.client.*` options. Secondary clusters can use independent authentication and TLS settings:
+
+```scala
+.option("pulsar.client.authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationToken")
+.option("pulsar.client.authParams", "token:<primary token>")
+.option("pulsar.client.tlsTrustCertsFilePath", "/path/to/primary-ca.pem")
+.option("pulsar.failover.secondary.0.authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationToken")
+.option("pulsar.failover.secondary.0.authParams", "token:<secondary token>")
+.option("pulsar.failover.secondary.0.tlsTrustCertsFilePath", "/path/to/secondary-ca.pem")
+.option("pulsar.failover.secondary.0.tlsTrustStorePath", "/path/to/secondary-truststore.jks")
+.option("pulsar.failover.secondary.0.tlsTrustStorePassword", "<password>")
+```
+
+Secondary indexes must start at `0` and be continuous. At least one secondary cluster is required; use plain `service.url` if no secondary cluster is needed. Only `ORDER` policy is supported now.
+
+Limitations:
+
+* `PulsarAdmin` does not participate in failover. If `maxBytesPerTrigger` requires `admin.url`, use DNS or a load balancer for admin high availability.
+* Pulsar `AutoClusterFailover` probes service endpoints at TCP level, so it may not detect partial broker-side degradation while proxies or ports remain reachable.
+
 ### Authentication
 Should the Pulsar cluster require authentication, credentials can be set in the following way.
 
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala b/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala
index 2cafa21..8d0af55 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala
@@ -40,16 +40,35 @@ private[pulsar] object CachedPulsarClient extends Logging {
 
   private val cacheLoader = new CacheLoader[ju.Map[String, Object], PulsarClientImpl]() {
     override def load(config: ju.Map[String, Object]): PulsarClientImpl = {
-      val pulsarServiceUrl = config.get(PulsarOptions.ServiceUrlOptionKey).toString
+      val failoverConfig = PulsarFailoverConfig.fromParams(config)
+      val pulsarServiceUrl = failoverConfig
+        .map(_.primaryServiceUrl)
+        .getOrElse(config.get(PulsarOptions.ServiceUrlOptionKey).toString)
+      val configWithoutFailover = config.asScala.toMap.filterNot { case (key, _) =>
+        key.toLowerCase(java.util.Locale.ROOT).startsWith(PulsarFailoverOptionKeyPrefix)
+      }
       val clientConf =
-        PulsarConfigUpdater("pulsarClientCache", config.asScala.toMap, PulsarOptions.FilteredKeys)
-          .rebuild()
+        PulsarConfigUpdater(
+          "pulsarClientCache",
+          configWithoutFailover,
+          PulsarOptions.FilteredKeys).rebuild()
 
       val builder = PulsarClient.builder()
       try {
-        builder
-          .loadConf(clientConf)
-          .serviceUrl(pulsarServiceUrl)
+        builder.loadConf(clientConf)
+        failoverConfig match {
+          case Some(failover) =>
+            val serviceUrlFromConfig = Option(config.get(PulsarOptions.ServiceUrlOptionKey))
+              .map(_.toString)
+            if (serviceUrlFromConfig.exists(_ != failover.primaryServiceUrl)) {
+              logWarning(
+                s"$ServiceUrlOptionKey differs from $PulsarFailoverPrimaryServiceUrlOptionKey; " +
+                  s"using ${failover.primaryServiceUrl} for Pulsar client failover")
+            }
+            builder.serviceUrlProvider(PulsarFailoverConfig.toServiceUrlProvider(failover))
+          case None =>
+            builder.serviceUrl(pulsarServiceUrl)
+        }
 
         // Set authentication parameters.
         if (clientConf.containsKey(AuthPluginClassName)) {
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
new file mode 100644
index 0000000..1557054
--- /dev/null
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
@@ -0,0 +1,308 @@
+/*
+ * Licensed 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.spark.sql.pulsar
+
+import java.time.Duration
+import java.util.Locale
+import java.util.concurrent.TimeUnit
+import java.{util => ju}
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.pulsar.client.api.{
+  AuthenticationFactory,
+  AutoClusterFailoverBuilder,
+  ServiceUrlProvider
+}
+import org.apache.pulsar.client.impl.AutoClusterFailover
+
+import org.apache.spark.sql.pulsar.PulsarOptions._
+
+private[pulsar] case class SecondaryClusterConfig(
+    serviceUrl: String,
+    auth: Option[(String, String)],
+    tlsTrustCertsFilePath: Option[String],
+    tlsTrustStorePath: Option[String],
+    tlsTrustStorePassword: Option[String])
+
+private[pulsar] case class PulsarFailoverConfig(
+    primaryServiceUrl: String,
+    secondaries: Seq[SecondaryClusterConfig],
+    failoverDelay: Duration,
+    switchBackDelay: Duration,
+    checkInterval: Duration,
+    policy: AutoClusterFailoverBuilder.FailoverPolicy)
+
+private[pulsar] object PulsarFailoverConfig {
+
+  private val DefaultFailoverDelayMs = 30000L
+  private val DefaultSwitchBackDelayMs = 60000L
+  private val DefaultCheckIntervalMs = 30000L
+
+  private val SecondaryServiceUrl = "serviceurl"
+  private val SecondaryAuthPluginClassName = "authpluginclassname"
+  private val SecondaryAuthParams = "authparams"
+  private val SecondaryTlsTrustCertsFilePath = "tlstrustcertsfilepath"
+  private val SecondaryTlsTrustStorePath = "tlstruststorepath"
+  private val SecondaryTlsTrustStorePassword = "tlstruststorepassword"
+
+  private val TopLevelKeys = Set(
+    PulsarFailoverPrimaryServiceUrlOptionKey,
+    PulsarFailoverDelayMsOptionKey,
+    PulsarFailoverSwitchBackDelayMsOptionKey,
+    PulsarFailoverCheckIntervalMsOptionKey,
+    PulsarFailoverPolicyOptionKey)
+
+  private val SecondaryKeys = Set(
+    SecondaryServiceUrl,
+    SecondaryAuthPluginClassName,
+    SecondaryAuthParams,
+    SecondaryTlsTrustCertsFilePath,
+    SecondaryTlsTrustStorePath,
+    SecondaryTlsTrustStorePassword)
+
+  def primaryServiceUrl(params: Map[String, String]): Option[String] = {
+    normalize(params).get(PulsarFailoverPrimaryServiceUrlOptionKey).map(_.trim).filter(_.nonEmpty)
+  }
+
+  /** Returns Some(...) if any pulsar.failover.* key is present, else None. */
+  def fromParams(params: ju.Map[String, Object]): Option[PulsarFailoverConfig] = {
+    fromParams(params.asScala.toMap.map { case (k, v) => k -> Option(v).map(_.toString).orNull })
+  }
+
+  /** Returns Some(...) if any pulsar.failover.* key is present, else None. */
+  def fromParams(params: Map[String, String]): Option[PulsarFailoverConfig] = {
+    val normalized = normalize(params)
+    val failoverParams = normalized.filter { case (k, _) =>
+      k.startsWith(PulsarFailoverOptionKeyPrefix)
+    }
+    if (failoverParams.isEmpty) {
+      return None
+    }
+
+    validateKnownKeys(failoverParams.keySet)
+
+    val primary = requiredNonEmpty(
+      failoverParams,
+      PulsarFailoverPrimaryServiceUrlOptionKey,
+      s"$PulsarFailoverPrimaryServiceUrlOptionKey must be specified when Pulsar failover " +
+        "is enabled")
+
+    val secondaries = parseSecondaries(failoverParams)
+    if (secondaries.isEmpty) {
+      throw new IllegalArgumentException(
+        s"${PulsarFailoverSecondaryDisplayPrefix}0.serviceUrl must be specified when Pulsar " +
+          s"failover is enabled; use $ServiceUrlOptionKey directly if no secondary cluster " +
+          "is needed")
+    }
+
+    Some(PulsarFailoverConfig(
+      primary,
+      secondaries,
+      parsePositiveDuration(
+        failoverParams,
+        PulsarFailoverDelayMsOptionKey,
+        DefaultFailoverDelayMs),
+      parsePositiveDuration(
+        failoverParams,
+        PulsarFailoverSwitchBackDelayMsOptionKey,
+        DefaultSwitchBackDelayMs),
+      parsePositiveDuration(
+        failoverParams,
+        PulsarFailoverCheckIntervalMsOptionKey,
+        DefaultCheckIntervalMs),
+      parsePolicy(failoverParams)))
+  }
+
+  /** Build a ServiceUrlProvider from this config. */
+  def toServiceUrlProvider(cfg: PulsarFailoverConfig): ServiceUrlProvider = {
+    val secondaryServiceUrls = cfg.secondaries.map(_.serviceUrl).asJava
+    val secondaryAuth = completeSecondaryMap[org.apache.pulsar.client.api.Authentication](
+      cfg.secondaries) { secondary =>
+      secondary.auth.map { case (pluginClassName, authParams) =>
+        AuthenticationFactory.create(pluginClassName, authParams)
+      }
+    }
+    val secondaryTlsTrustCertsFilePath = completeSecondaryMap[String](cfg.secondaries)(
+      _.tlsTrustCertsFilePath)
+    val secondaryTlsTrustStorePath = completeSecondaryMap[String](cfg.secondaries)(
+      _.tlsTrustStorePath)
+    val secondaryTlsTrustStorePassword = completeSecondaryMap[String](cfg.secondaries)(
+      _.tlsTrustStorePassword)
+
+    val builder = AutoClusterFailover
+      .builder()
+      .primary(cfg.primaryServiceUrl)
+      .secondary(secondaryServiceUrls)
+      .failoverPolicy(cfg.policy)
+      .failoverDelay(cfg.failoverDelay.toMillis, TimeUnit.MILLISECONDS)
+      .switchBackDelay(cfg.switchBackDelay.toMillis, TimeUnit.MILLISECONDS)
+      .checkInterval(cfg.checkInterval.toMillis, TimeUnit.MILLISECONDS)
+    secondaryAuth.foreach(auth => builder.secondaryAuthentication(auth.asJava))
+    secondaryTlsTrustCertsFilePath.foreach(tls =>
+      builder.secondaryTlsTrustCertsFilePath(tls.asJava))
+    secondaryTlsTrustStorePath.foreach(tls => builder.secondaryTlsTrustStorePath(tls.asJava))
+    secondaryTlsTrustStorePassword.foreach(tls =>
+      builder.secondaryTlsTrustStorePassword(tls.asJava))
+    builder.build()
+  }
+
+  private def completeSecondaryMap[T >: Null](
+      secondaries: Seq[SecondaryClusterConfig])(
+      value: SecondaryClusterConfig => Option[T]): Option[Map[String, T]] = {
+    val values = secondaries.map(secondary => secondary.serviceUrl -> value(secondary))
+    if (values.exists(_._2.isDefined)) {
+      Some(values.map { case (serviceUrl, maybeValue) => serviceUrl -> maybeValue.orNull }.toMap)
+    } else {
+      None
+    }
+  }
+
+  private def normalize(params: Map[String, String]): Map[String, String] = {
+    params.map { case (k, v) => normalizeKey(k) -> v }
+  }
+
+  private def normalizeKey(key: String): String = {
+    val lower = key.toLowerCase(Locale.ROOT)
+    if (lower.startsWith(PulsarFailoverOptionKeyPrefix)) {
+      PulsarFailoverOptionKeyPrefix +
+        lower.substring(PulsarFailoverOptionKeyPrefix.length).replace(".", "")
+    } else {
+      lower.replace(".", "")
+    }
+  }
+
+  private def validateKnownKeys(keys: Set[String]): Unit = {
+    keys.foreach {
+      case key if TopLevelKeys.contains(key) =>
+      case key if parseSecondaryKey(key).exists { case (_, name) =>
+        SecondaryKeys.contains(name)
+      } =>
+      case key =>
+        throw new IllegalArgumentException(s"Unsupported Pulsar failover option: $key")
+    }
+  }
+
+  private def parseSecondaryKey(key: String): Option[(Int, String)] = {
+    val normalizedPrefix = normalizeKey(PulsarFailoverSecondaryPrefix)
+    if (!key.startsWith(normalizedPrefix)) {
+      return None
+    }
+    val rest = key.substring(normalizedPrefix.length)
+    val index = rest.takeWhile(_.isDigit)
+    val name = rest.drop(index.length)
+    if (index.nonEmpty && name.nonEmpty) {
+      Some(index.toInt -> name.stripPrefix("."))
+    } else {
+      None
+    }
+  }
+
+  private def parseSecondaries(params: Map[String, String]): Seq[SecondaryClusterConfig] = {
+    val grouped = params.toSeq.flatMap { case (key, value) =>
+      parseSecondaryKey(key).map { case (index, name) => index -> (name -> value) }
+    }.groupBy(_._1).map { case (index, entries) =>
+      index -> entries.map(_._2).toMap
+    }
+
+    if (grouped.isEmpty) {
+      return Seq.empty
+    }
+
+    val indexes = grouped.keys.toSeq.sorted
+    val expected = indexes.head to indexes.last
+    if (indexes.head != 0 || indexes != expected) {
+      val missing = expected.filterNot(grouped.contains)
+      throw new IllegalArgumentException(
+        s"Pulsar failover secondary indexes must start at 0 and be continuous; " +
+          s"configured indexes: ${indexes.mkString(",")}, missing indexes: " +
+          missing.mkString(","))
+    }
+
+    indexes.map { index =>
+      val secondaryParams = grouped(index)
+      val serviceUrl = requiredNonEmpty(
+        secondaryParams,
+        SecondaryServiceUrl,
+        s"${PulsarFailoverSecondaryDisplayPrefix}$index.serviceUrl must be specified")
+      val auth = parseAuth(secondaryParams, index)
+      SecondaryClusterConfig(
+        serviceUrl,
+        auth,
+        nonEmpty(secondaryParams, SecondaryTlsTrustCertsFilePath),
+        nonEmpty(secondaryParams, SecondaryTlsTrustStorePath),
+        nonEmpty(secondaryParams, SecondaryTlsTrustStorePassword))
+    }
+  }
+
+  private def parseAuth(params: Map[String, String], index: Int): Option[(String, String)] = {
+    val pluginClassName = nonEmpty(params, SecondaryAuthPluginClassName)
+    val authParams = nonEmpty(params, SecondaryAuthParams)
+    (pluginClassName, authParams) match {
+      case (Some(pluginClassName), Some(authParams)) => Some(pluginClassName -> authParams)
+      case (None, None) => None
+      case _ =>
+        throw new IllegalArgumentException(
+          s"${PulsarFailoverSecondaryDisplayPrefix}$index.authPluginClassName and " +
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.authParams must be specified " +
+            "together")
+    }
+  }
+
+  private def parsePositiveDuration(
+      params: Map[String, String],
+      key: String,
+      defaultMs: Long): Duration = {
+    val millis = params.get(key).map(_.trim).filter(_.nonEmpty) match {
+      case Some(value) =>
+        try {
+          value.toLong
+        } catch {
+          case _: NumberFormatException =>
+            throw new IllegalArgumentException(
+              s"$key must be a positive milliseconds value: $value")
+        }
+      case None => defaultMs
+    }
+    if (millis <= 0) {
+      throw new IllegalArgumentException(s"$key must be positive, but was $millis")
+    }
+    Duration.ofMillis(millis)
+  }
+
+  private def parsePolicy(
+      params: Map[String, String]): AutoClusterFailoverBuilder.FailoverPolicy = {
+    params.get(PulsarFailoverPolicyOptionKey).map(_.trim).filter(_.nonEmpty) match {
+      case Some(policy) =>
+        try {
+          AutoClusterFailoverBuilder.FailoverPolicy.valueOf(policy.toUpperCase(Locale.ROOT))
+        } catch {
+          case _: IllegalArgumentException =>
+            throw new IllegalArgumentException(s"Unsupported Pulsar failover policy: $policy")
+        }
+      case None => AutoClusterFailoverBuilder.FailoverPolicy.ORDER
+    }
+  }
+
+  private def requiredNonEmpty(
+      params: Map[String, String],
+      key: String,
+      message: String): String = {
+    nonEmpty(params, key).getOrElse(throw new IllegalArgumentException(message))
+  }
+
+  private def nonEmpty(params: Map[String, String], key: String): Option[String] = {
+    params.get(key).map(_.trim).filter(_.nonEmpty)
+  }
+}
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala
index 4b7ac1c..dee7747 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala
@@ -25,6 +25,7 @@ private[pulsar] object PulsarOptions {
   val PulsarAdminOptionKeyPrefix: String = "pulsar.admin."
   val PulsarProducerOptionKeyPrefix: String = "pulsar.producer."
   val PulsarReaderOptionKeyPrefix: String = "pulsar.reader."
+  val PulsarFailoverOptionKeyPrefix: String = "pulsar.failover."
 
   // options
 
@@ -38,6 +39,18 @@ private[pulsar] object PulsarOptions {
 
   val ServiceUrlOptionKey: String = "service.url"
   val AdminUrlOptionKey: String = "admin.url"
+
+  val PulsarFailoverPrimaryServiceUrlOptionKey: String =
+    s"${PulsarFailoverOptionKeyPrefix}primaryserviceurl"
+  val PulsarFailoverSecondaryPrefix: String = s"${PulsarFailoverOptionKeyPrefix}secondary."
+  val PulsarFailoverSecondaryDisplayPrefix: String = PulsarFailoverSecondaryPrefix
+  val PulsarFailoverDelayMsOptionKey: String = s"${PulsarFailoverOptionKeyPrefix}failoverdelayms"
+  val PulsarFailoverSwitchBackDelayMsOptionKey: String =
+    s"${PulsarFailoverOptionKeyPrefix}switchbackdelayms"
+  val PulsarFailoverCheckIntervalMsOptionKey: String =
+    s"${PulsarFailoverOptionKeyPrefix}checkintervalms"
+  val PulsarFailoverPolicyOptionKey: String = s"${PulsarFailoverOptionKeyPrefix}policy"
+
   val StartingOffsetsOptionKey: String = "startingOffsets".toLowerCase(Locale.ROOT)
   val StartingTime: String = "startingTime".toLowerCase(Locale.ROOT)
   val EndingTime: String = "endingTime".toLowerCase(Locale.ROOT)
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
index 998f951..bc2aceb 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
@@ -267,6 +267,12 @@ private[pulsar] object PulsarProvider extends Logging {
     }
   }
 
+  private def getFailoverParams(parameters: Map[String, String]): Map[String, String] = {
+    parameters.filter { case (k, _) =>
+      k.toLowerCase(Locale.ROOT).startsWith(PulsarFailoverOptionKeyPrefix)
+    }
+  }
+
   private def getAdminParams(parameters: Map[String, String]): Map[String, String] = {
     getModuleParams(parameters, PulsarAdminOptionKeyPrefix, clientConfKeys)
   }
@@ -386,7 +392,8 @@ private[pulsar] object PulsarProvider extends Logging {
   }
 
   private def getServiceUrl(parameters: Map[String, String]): String = {
-    parameters(ServiceUrlOptionKey)
+    PulsarFailoverConfig.primaryServiceUrl(parameters)
+      .getOrElse(parameters(ServiceUrlOptionKey))
   }
 
   private def getAdminUrl(parameters: Map[String, String]): Option[String] = {
@@ -416,8 +423,19 @@ private[pulsar] object PulsarProvider extends Logging {
 
   private def validateGeneralOptions(
       caseInsensitiveParams: Map[String, String]): Map[String, String] = {
-    if (!caseInsensitiveParams.contains(ServiceUrlOptionKey)) {
-      throw new IllegalArgumentException(s"$ServiceUrlOptionKey must be specified")
+    val failoverConfig = PulsarFailoverConfig.fromParams(caseInsensitiveParams)
+    failoverConfig match {
+      case Some(config) =>
+        caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
+          require(
+            serviceUrl == config.primaryServiceUrl,
+            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlOptionKey " +
+              "when Pulsar failover is enabled")
+        }
+      case None =>
+        require(
+          caseInsensitiveParams.contains(ServiceUrlOptionKey),
+          s"$ServiceUrlOptionKey must be specified")
     }
 
     // validate topic options
@@ -503,8 +521,19 @@ private[pulsar] object PulsarProvider extends Logging {
   private def validateSinkOptions(parameters: Map[String, String]): Map[String, String] = {
     val caseInsensitiveParams = parameters.map { case (k, v) => (k.toLowerCase(Locale.ROOT), v) }
 
-    if (!caseInsensitiveParams.contains(ServiceUrlOptionKey)) {
-      throw new IllegalArgumentException(s"$ServiceUrlOptionKey must be specified")
+    val failoverConfig = PulsarFailoverConfig.fromParams(caseInsensitiveParams)
+    failoverConfig match {
+      case Some(config) =>
+        caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
+          require(
+            serviceUrl == config.primaryServiceUrl,
+            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlOptionKey " +
+              "when Pulsar failover is enabled")
+        }
+      case None =>
+        require(
+          caseInsensitiveParams.contains(ServiceUrlOptionKey),
+          s"$ServiceUrlOptionKey must be specified")
     }
 
     val topicOptions =
@@ -525,7 +554,7 @@ private[pulsar] object PulsarProvider extends Logging {
 
     val serviceUrl = getServiceUrl(parameters)
     val adminUrl = getAdminUrl(parameters)
-    var clientParams = getClientParams(parameters)
+    var clientParams = getClientParams(parameters) ++ getFailoverParams(parameters)
     clientParams += (ServiceUrlOptionKey -> serviceUrl)
     val readerParams = getReaderParams(parameters)
     val adminParams = getAdminParams(parameters)
@@ -542,7 +571,7 @@ private[pulsar] object PulsarProvider extends Logging {
 
     val serviceUrl = getServiceUrl(parameters)
 
-    var clientParams = getClientParams(parameters)
+    var clientParams = getClientParams(parameters) ++ getFailoverParams(parameters)
     clientParams += (ServiceUrlOptionKey -> serviceUrl)
     val producerParams = getProducerParams(parameters)
 
diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
new file mode 100644
index 0000000..0b76c9a
--- /dev/null
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
@@ -0,0 +1,132 @@
+/*
+ * Licensed 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.spark.sql.pulsar
+
+import java.time.Duration
+
+import org.apache.pulsar.client.api.AutoClusterFailoverBuilder.FailoverPolicy
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.pulsar.PulsarOptions._
+
+class PulsarFailoverConfigSuite extends SparkFunSuite {
+
+  private val primaryUrl = "pulsar://primary:6650"
+  private val secondary0Url = "pulsar://secondary-0:6650"
+  private val secondary1Url = "pulsar://secondary-1:6650"
+
+  test("fromParams returns None when no failover option is configured") {
+    assert(PulsarFailoverConfig.fromParams(Map(ServiceUrlOptionKey -> primaryUrl)).isEmpty)
+  }
+
+  test("missing secondary.0.serviceUrl fails with guidance to use service.url") {
+    val error = intercept[IllegalArgumentException] {
+      PulsarFailoverConfig.fromParams(Map(PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl))
+    }
+
+    assert(error.getMessage.contains("secondary.0.serviceUrl"))
+    assert(error.getMessage.contains(ServiceUrlOptionKey))
+  }
+
+  test("secondary indexes must start from 0 and be continuous") {
+    val error = intercept[IllegalArgumentException] {
+      PulsarFailoverConfig.fromParams(Map(
+        PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+        s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+        s"${PulsarFailoverSecondaryPrefix}2.serviceUrl" -> secondary1Url))
+    }
+
+    assert(error.getMessage.contains("continuous"))
+    assert(error.getMessage.contains("missing indexes: 1"))
+  }
+
+  test("non-positive duration fails") {
+    val error = intercept[IllegalArgumentException] {
+      PulsarFailoverConfig.fromParams(Map(
+        PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+        s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+        PulsarFailoverDelayMsOptionKey -> "0"))
+    }
+
+    assert(error.getMessage.contains(PulsarFailoverDelayMsOptionKey))
+    assert(error.getMessage.contains("positive"))
+  }
+
+  test("full config parses durations, policy, auth and tls per secondary") {
+    val config = PulsarFailoverConfig.fromParams(Map(
+      PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+      PulsarFailoverDelayMsOptionKey -> "5000",
+      PulsarFailoverSwitchBackDelayMsOptionKey -> "10000",
+      PulsarFailoverCheckIntervalMsOptionKey -> "15000",
+      PulsarFailoverPolicyOptionKey -> "order",
+      s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+      s"${PulsarFailoverSecondaryPrefix}0.authPluginClassName" -> "plugin0",
+      s"${PulsarFailoverSecondaryPrefix}0.authParams" -> "params0",
+      s"${PulsarFailoverSecondaryPrefix}0.tlsTrustCertsFilePath" -> "/cert0.pem",
+      s"${PulsarFailoverSecondaryPrefix}0.tlsTrustStorePath" -> "/truststore0.jks",
+      s"${PulsarFailoverSecondaryPrefix}0.tlsTrustStorePassword" -> "password0",
+      s"${PulsarFailoverSecondaryPrefix}1.serviceUrl" -> secondary1Url,
+      s"${PulsarFailoverSecondaryPrefix}1.authPluginClassName" -> "plugin1",
+      s"${PulsarFailoverSecondaryPrefix}1.authParams" -> "params1",
+      s"${PulsarFailoverSecondaryPrefix}1.tlsTrustCertsFilePath" -> "/cert1.pem",
+      s"${PulsarFailoverSecondaryPrefix}1.tlsTrustStorePath" -> "/truststore1.jks",
+      s"${PulsarFailoverSecondaryPrefix}1.tlsTrustStorePassword" -> "password1")).get
+
+    assert(config.primaryServiceUrl === primaryUrl)
+    assert(config.failoverDelay === Duration.ofMillis(5000))
+    assert(config.switchBackDelay === Duration.ofMillis(10000))
+    assert(config.checkInterval === Duration.ofMillis(15000))
+    assert(config.policy === FailoverPolicy.ORDER)
+    assert(config.secondaries === Seq(
+      SecondaryClusterConfig(
+        secondary0Url,
+        Some("plugin0" -> "params0"),
+        Some("/cert0.pem"),
+        Some("/truststore0.jks"),
+        Some("password0")),
+      SecondaryClusterConfig(
+        secondary1Url,
+        Some("plugin1" -> "params1"),
+        Some("/cert1.pem"),
+        Some("/truststore1.jks"),
+        Some("password1"))))
+  }
+
+  test("secondary auth plugin and params must be configured together") {
+    val error = intercept[IllegalArgumentException] {
+      PulsarFailoverConfig.fromParams(Map(
+        PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+        s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+        s"${PulsarFailoverSecondaryPrefix}0.authPluginClassName" -> "plugin0"))
+    }
+
+    assert(error.getMessage.contains("authPluginClassName"))
+    assert(error.getMessage.contains("authParams"))
+  }
+
+  test("toServiceUrlProvider builds AutoClusterFailover provider") {
+    val config = PulsarFailoverConfig(
+      primaryUrl,
+      Seq(SecondaryClusterConfig(secondary0Url, None, None, None, None)),
+      Duration.ofMillis(5000),
+      Duration.ofMillis(10000),
+      Duration.ofMillis(15000),
+      FailoverPolicy.ORDER)
+
+    val provider = PulsarFailoverConfig.toServiceUrlProvider(config)
+
+    assert(provider.getClass.getName.contains("AutoClusterFailover"))
+    assert(provider.getServiceUrl === primaryUrl)
+  }
+}

From fbd7e5278ae2c602a6762edc4495e6a63fc61ee6 Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Thu, 14 May 2026 15:54:37 +0800
Subject: [PATCH 2/7] Add Pulsar auto cluster failover support

---
 .../spark/sql/pulsar/CachedPulsarClient.scala |   6 +-
 .../sql/pulsar/PulsarFailoverConfig.scala     | 119 +++++++++++-------
 .../spark/sql/pulsar/PulsarOptions.scala      |   7 ++
 .../spark/sql/pulsar/PulsarProvider.scala     |  42 ++++---
 .../pulsar/PulsarFailoverConfigSuite.scala    | 114 ++++++++++-------
 5 files changed, 179 insertions(+), 109 deletions(-)

diff --git a/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala b/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala
index 8d0af55..e27f2af 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/CachedPulsarClient.scala
@@ -45,7 +45,9 @@ private[pulsar] object CachedPulsarClient extends Logging {
         .map(_.primaryServiceUrl)
         .getOrElse(config.get(PulsarOptions.ServiceUrlOptionKey).toString)
       val configWithoutFailover = config.asScala.toMap.filterNot { case (key, _) =>
-        key.toLowerCase(java.util.Locale.ROOT).startsWith(PulsarFailoverOptionKeyPrefix)
+        val normalizedKey = key.toLowerCase(java.util.Locale.ROOT)
+        normalizedKey.startsWith(PulsarFailoverOptionKeyPrefix) ||
+        failoverConfig.isDefined && normalizedKey == ServiceUrlOptionKey
       }
       val clientConf =
         PulsarConfigUpdater(
@@ -62,7 +64,7 @@ private[pulsar] object CachedPulsarClient extends Logging {
               .map(_.toString)
             if (serviceUrlFromConfig.exists(_ != failover.primaryServiceUrl)) {
               logWarning(
-                s"$ServiceUrlOptionKey differs from $PulsarFailoverPrimaryServiceUrlOptionKey; " +
+                s"$ServiceUrlOptionKey differs from $PulsarFailoverPrimaryServiceUrlDisplayKey; " +
                   s"using ${failover.primaryServiceUrl} for Pulsar client failover")
             }
             builder.serviceUrlProvider(PulsarFailoverConfig.toServiceUrlProvider(failover))
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
index 1557054..1e5cf24 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
@@ -78,7 +78,7 @@ private[pulsar] object PulsarFailoverConfig {
 
   /** Returns Some(...) if any pulsar.failover.* key is present, else None. */
   def fromParams(params: ju.Map[String, Object]): Option[PulsarFailoverConfig] = {
-    fromParams(params.asScala.toMap.map { case (k, v) => k -> Option(v).map(_.toString).orNull })
+    fromParams(params.asScala.toMap.collect { case (k, v) if v != null => k -> v.toString })
   }
 
   /** Returns Some(...) if any pulsar.failover.* key is present, else None. */
@@ -93,10 +93,14 @@ private[pulsar] object PulsarFailoverConfig {
 
     validateKnownKeys(failoverParams.keySet)
 
+    if (!failoverParams.contains(PulsarFailoverPrimaryServiceUrlOptionKey)) {
+      return None
+    }
+
     val primary = requiredNonEmpty(
       failoverParams,
       PulsarFailoverPrimaryServiceUrlOptionKey,
-      s"$PulsarFailoverPrimaryServiceUrlOptionKey must be specified when Pulsar failover " +
+      s"$PulsarFailoverPrimaryServiceUrlDisplayKey must be specified when Pulsar failover " +
         "is enabled")
 
     val secondaries = parseSecondaries(failoverParams)
@@ -107,39 +111,41 @@ private[pulsar] object PulsarFailoverConfig {
           "is needed")
     }
 
-    Some(PulsarFailoverConfig(
-      primary,
-      secondaries,
-      parsePositiveDuration(
-        failoverParams,
-        PulsarFailoverDelayMsOptionKey,
-        DefaultFailoverDelayMs),
-      parsePositiveDuration(
-        failoverParams,
-        PulsarFailoverSwitchBackDelayMsOptionKey,
-        DefaultSwitchBackDelayMs),
-      parsePositiveDuration(
-        failoverParams,
-        PulsarFailoverCheckIntervalMsOptionKey,
-        DefaultCheckIntervalMs),
-      parsePolicy(failoverParams)))
+    Some(
+      PulsarFailoverConfig(
+        primary,
+        secondaries,
+        parsePositiveDuration(
+          failoverParams,
+          PulsarFailoverDelayMsOptionKey,
+          DefaultFailoverDelayMs),
+        parsePositiveDuration(
+          failoverParams,
+          PulsarFailoverSwitchBackDelayMsOptionKey,
+          DefaultSwitchBackDelayMs),
+        parsePositiveDuration(
+          failoverParams,
+          PulsarFailoverCheckIntervalMsOptionKey,
+          DefaultCheckIntervalMs),
+        parsePolicy(failoverParams)))
   }
 
   /** Build a ServiceUrlProvider from this config. */
   def toServiceUrlProvider(cfg: PulsarFailoverConfig): ServiceUrlProvider = {
     val secondaryServiceUrls = cfg.secondaries.map(_.serviceUrl).asJava
-    val secondaryAuth = completeSecondaryMap[org.apache.pulsar.client.api.Authentication](
-      cfg.secondaries) { secondary =>
-      secondary.auth.map { case (pluginClassName, authParams) =>
-        AuthenticationFactory.create(pluginClassName, authParams)
+    val secondaryAuth =
+      completeSecondaryMap[org.apache.pulsar.client.api.Authentication](cfg.secondaries) {
+        secondary =>
+          secondary.auth.map { case (pluginClassName, authParams) =>
+            AuthenticationFactory.create(pluginClassName, authParams)
+          }
       }
-    }
-    val secondaryTlsTrustCertsFilePath = completeSecondaryMap[String](cfg.secondaries)(
-      _.tlsTrustCertsFilePath)
-    val secondaryTlsTrustStorePath = completeSecondaryMap[String](cfg.secondaries)(
-      _.tlsTrustStorePath)
-    val secondaryTlsTrustStorePassword = completeSecondaryMap[String](cfg.secondaries)(
-      _.tlsTrustStorePassword)
+    val secondaryTlsTrustCertsFilePath =
+      completeSecondaryMap[String](cfg.secondaries)(_.tlsTrustCertsFilePath)
+    val secondaryTlsTrustStorePath =
+      completeSecondaryMap[String](cfg.secondaries)(_.tlsTrustStorePath)
+    val secondaryTlsTrustStorePassword =
+      completeSecondaryMap[String](cfg.secondaries)(_.tlsTrustStorePassword)
 
     val builder = AutoClusterFailover
       .builder()
@@ -158,8 +164,7 @@ private[pulsar] object PulsarFailoverConfig {
     builder.build()
   }
 
-  private def completeSecondaryMap[T >: Null](
-      secondaries: Seq[SecondaryClusterConfig])(
+  private def completeSecondaryMap[T >: Null](secondaries: Seq[SecondaryClusterConfig])(
       value: SecondaryClusterConfig => Option[T]): Option[Map[String, T]] = {
     val values = secondaries.map(secondary => secondary.serviceUrl -> value(secondary))
     if (values.exists(_._2.isDefined)) {
@@ -187,10 +192,11 @@ private[pulsar] object PulsarFailoverConfig {
     keys.foreach {
       case key if TopLevelKeys.contains(key) =>
       case key if parseSecondaryKey(key).exists { case (_, name) =>
-        SecondaryKeys.contains(name)
-      } =>
+            SecondaryKeys.contains(name)
+          } =>
       case key =>
-        throw new IllegalArgumentException(s"Unsupported Pulsar failover option: $key")
+        throw new IllegalArgumentException(
+          s"Unsupported Pulsar failover option: ${displayKey(key)}")
     }
   }
 
@@ -210,11 +216,14 @@ private[pulsar] object PulsarFailoverConfig {
   }
 
   private def parseSecondaries(params: Map[String, String]): Seq[SecondaryClusterConfig] = {
-    val grouped = params.toSeq.flatMap { case (key, value) =>
-      parseSecondaryKey(key).map { case (index, name) => index -> (name -> value) }
-    }.groupBy(_._1).map { case (index, entries) =>
-      index -> entries.map(_._2).toMap
-    }
+    val grouped = params.toSeq
+      .flatMap { case (key, value) =>
+        parseSecondaryKey(key).map { case (index, name) => index -> (name -> value) }
+      }
+      .groupBy(_._1)
+      .map { case (index, entries) =>
+        index -> entries.map(_._2).toMap
+      }
 
     if (grouped.isEmpty) {
       return Seq.empty
@@ -227,7 +236,7 @@ private[pulsar] object PulsarFailoverConfig {
       throw new IllegalArgumentException(
         s"Pulsar failover secondary indexes must start at 0 and be continuous; " +
           s"configured indexes: ${indexes.mkString(",")}, missing indexes: " +
-          missing.mkString(","))
+          (if (missing.nonEmpty) missing.mkString(",") else "before 0"))
     }
 
     indexes.map { index =>
@@ -271,12 +280,12 @@ private[pulsar] object PulsarFailoverConfig {
         } catch {
           case _: NumberFormatException =>
             throw new IllegalArgumentException(
-              s"$key must be a positive milliseconds value: $value")
+              s"${displayKey(key)} must be a positive milliseconds value: $value")
         }
       case None => defaultMs
     }
     if (millis <= 0) {
-      throw new IllegalArgumentException(s"$key must be positive, but was $millis")
+      throw new IllegalArgumentException(s"${displayKey(key)} must be positive, but was $millis")
     }
     Duration.ofMillis(millis)
   }
@@ -305,4 +314,30 @@ private[pulsar] object PulsarFailoverConfig {
   private def nonEmpty(params: Map[String, String], key: String): Option[String] = {
     params.get(key).map(_.trim).filter(_.nonEmpty)
   }
+
+  private def displayKey(key: String): String = {
+    key match {
+      case PulsarFailoverPrimaryServiceUrlOptionKey => PulsarFailoverPrimaryServiceUrlDisplayKey
+      case PulsarFailoverDelayMsOptionKey => PulsarFailoverDelayMsDisplayKey
+      case PulsarFailoverSwitchBackDelayMsOptionKey => PulsarFailoverSwitchBackDelayMsDisplayKey
+      case PulsarFailoverCheckIntervalMsOptionKey => PulsarFailoverCheckIntervalMsDisplayKey
+      case PulsarFailoverPolicyOptionKey => PulsarFailoverPolicyOptionKey
+      case secondaryKey =>
+        parseSecondaryKey(secondaryKey) match {
+          case Some((index, SecondaryServiceUrl)) =>
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.serviceUrl"
+          case Some((index, SecondaryAuthPluginClassName)) =>
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.authPluginClassName"
+          case Some((index, SecondaryAuthParams)) =>
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.authParams"
+          case Some((index, SecondaryTlsTrustCertsFilePath)) =>
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.tlsTrustCertsFilePath"
+          case Some((index, SecondaryTlsTrustStorePath)) =>
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.tlsTrustStorePath"
+          case Some((index, SecondaryTlsTrustStorePassword)) =>
+            s"${PulsarFailoverSecondaryDisplayPrefix}$index.tlsTrustStorePassword"
+          case _ => secondaryKey
+        }
+    }
+  }
 }
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala
index dee7747..bab9d4c 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarOptions.scala
@@ -40,15 +40,22 @@ private[pulsar] object PulsarOptions {
   val ServiceUrlOptionKey: String = "service.url"
   val AdminUrlOptionKey: String = "admin.url"
 
+  val PulsarFailoverPrimaryServiceUrlDisplayKey: String =
+    s"${PulsarFailoverOptionKeyPrefix}primary.serviceUrl"
   val PulsarFailoverPrimaryServiceUrlOptionKey: String =
     s"${PulsarFailoverOptionKeyPrefix}primaryserviceurl"
   val PulsarFailoverSecondaryPrefix: String = s"${PulsarFailoverOptionKeyPrefix}secondary."
   val PulsarFailoverSecondaryDisplayPrefix: String = PulsarFailoverSecondaryPrefix
   val PulsarFailoverDelayMsOptionKey: String = s"${PulsarFailoverOptionKeyPrefix}failoverdelayms"
+  val PulsarFailoverDelayMsDisplayKey: String = s"${PulsarFailoverOptionKeyPrefix}failoverDelayMs"
   val PulsarFailoverSwitchBackDelayMsOptionKey: String =
     s"${PulsarFailoverOptionKeyPrefix}switchbackdelayms"
+  val PulsarFailoverSwitchBackDelayMsDisplayKey: String =
+    s"${PulsarFailoverOptionKeyPrefix}switchBackDelayMs"
   val PulsarFailoverCheckIntervalMsOptionKey: String =
     s"${PulsarFailoverOptionKeyPrefix}checkintervalms"
+  val PulsarFailoverCheckIntervalMsDisplayKey: String =
+    s"${PulsarFailoverOptionKeyPrefix}checkIntervalMs"
   val PulsarFailoverPolicyOptionKey: String = s"${PulsarFailoverOptionKeyPrefix}policy"
 
   val StartingOffsetsOptionKey: String = "startingOffsets".toLowerCase(Locale.ROOT)
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
index bc2aceb..5d69ac2 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
@@ -57,8 +57,8 @@ private[pulsar] class PulsarProvider
       parameters: Map[String, String]): (String, StructType) = {
 
     val caseInsensitiveParams = validateStreamOptions(parameters)
-    val (clientConfig, _, adminConfig,
-      serviceUrlConfig, adminUrl) = prepareConfForReader(parameters)
+    val (clientConfig, _, adminConfig, serviceUrlConfig, adminUrl) = prepareConfForReader(
+      parameters)
 
     val subscriptionNamePrefix = s"spark-pulsar-${UUID.randomUUID}"
     val inferredSchema = Utils.tryWithResource(
@@ -89,8 +89,8 @@ private[pulsar] class PulsarProvider
     logDebug(s"Creating Pulsar source: $parameters")
 
     val caseInsensitiveParams = validateStreamOptions(parameters)
-    val (clientConfig, readerConfig,
-      adminConfig, serviceUrl, adminUrl) = prepareConfForReader(parameters)
+    val (clientConfig, readerConfig, adminConfig, serviceUrl, adminUrl) = prepareConfForReader(
+      parameters)
 
     logDebug(
       s"Client config: $clientConfig; Reader config: $readerConfig; Service URL: $serviceUrl")
@@ -117,8 +117,9 @@ private[pulsar] class PulsarProvider
 
     val maxBytes = maxBytesPerTrigger(caseInsensitiveParams)
     if (adminUrl.isEmpty && maxBytes != 0L) {
-      throw new IllegalArgumentException("admin.url " +
-        "must be specified if maxBytesPerTrigger is specified")
+      throw new IllegalArgumentException(
+        "admin.url " +
+          "must be specified if maxBytesPerTrigger is specified")
     }
 
     new PulsarSource(
@@ -142,8 +143,8 @@ private[pulsar] class PulsarProvider
 
     val subscriptionNamePrefix = getSubscriptionPrefix(parameters, isBatch = true)
 
-    val (clientConfig, readerConfig,
-      adminConfig, serviceUrl, adminUrl) = prepareConfForReader(parameters)
+    val (clientConfig, readerConfig, adminConfig, serviceUrl, adminUrl) = prepareConfForReader(
+      parameters)
 
     val (start, end, schema, pSchema) = Utils.tryWithResource(
       PulsarHelper(
@@ -392,7 +393,8 @@ private[pulsar] object PulsarProvider extends Logging {
   }
 
   private def getServiceUrl(parameters: Map[String, String]): String = {
-    PulsarFailoverConfig.primaryServiceUrl(parameters)
+    PulsarFailoverConfig
+      .primaryServiceUrl(parameters)
       .getOrElse(parameters(ServiceUrlOptionKey))
   }
 
@@ -416,10 +418,8 @@ private[pulsar] object PulsarProvider extends Logging {
 
   private def maxBytesPerTrigger(caseInsensitiveParams: Map[String, String]): Long =
     caseInsensitiveParams
-      .getOrElse(
-        PulsarOptions.MaxBytesPerTrigger,
-        0L.toString
-      ).toLong
+      .getOrElse(PulsarOptions.MaxBytesPerTrigger, 0L.toString)
+      .toLong
 
   private def validateGeneralOptions(
       caseInsensitiveParams: Map[String, String]): Map[String, String] = {
@@ -429,7 +429,7 @@ private[pulsar] object PulsarProvider extends Logging {
         caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
           require(
             serviceUrl == config.primaryServiceUrl,
-            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlOptionKey " +
+            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlDisplayKey " +
               "when Pulsar failover is enabled")
         }
       case None =>
@@ -527,7 +527,7 @@ private[pulsar] object PulsarProvider extends Logging {
         caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
           require(
             serviceUrl == config.primaryServiceUrl,
-            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlOptionKey " +
+            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlDisplayKey " +
               "when Pulsar failover is enabled")
         }
       case None =>
@@ -548,9 +548,12 @@ private[pulsar] object PulsarProvider extends Logging {
     caseInsensitiveParams
   }
 
-  private def prepareConfForReader(parameters: Map[String, String])
-      : (ju.Map[String, Object], ju.Map[String, Object],
-        ju.Map[String, Object], String, Option[String]) = {
+  private def prepareConfForReader(parameters: Map[String, String]): (
+      ju.Map[String, Object],
+      ju.Map[String, Object],
+      ju.Map[String, Object],
+      String,
+      Option[String]) = {
 
     val serviceUrl = getServiceUrl(parameters)
     val adminUrl = getAdminUrl(parameters)
@@ -563,7 +566,8 @@ private[pulsar] object PulsarProvider extends Logging {
       paramsToPulsarConf("pulsar.client", clientParams),
       paramsToPulsarConf("pulsar.reader", readerParams),
       paramsToPulsarConf("pulsar.admin", adminParams),
-      serviceUrl, adminUrl)
+      serviceUrl,
+      adminUrl)
   }
 
   private def prepareConfForProducer(parameters: Map[String, String])
diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
index 0b76c9a..1f8fb4f 100644
--- a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
@@ -15,6 +15,8 @@ package org.apache.spark.sql.pulsar
 
 import java.time.Duration
 
+import scala.jdk.CollectionConverters._
+
 import org.apache.pulsar.client.api.AutoClusterFailoverBuilder.FailoverPolicy
 
 import org.apache.spark.SparkFunSuite
@@ -41,10 +43,11 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
 
   test("secondary indexes must start from 0 and be continuous") {
     val error = intercept[IllegalArgumentException] {
-      PulsarFailoverConfig.fromParams(Map(
-        PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
-        s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
-        s"${PulsarFailoverSecondaryPrefix}2.serviceUrl" -> secondary1Url))
+      PulsarFailoverConfig.fromParams(
+        Map(
+          PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+          s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+          s"${PulsarFailoverSecondaryPrefix}2.serviceUrl" -> secondary1Url))
     }
 
     assert(error.getMessage.contains("continuous"))
@@ -53,68 +56,82 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
 
   test("non-positive duration fails") {
     val error = intercept[IllegalArgumentException] {
-      PulsarFailoverConfig.fromParams(Map(
-        PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
-        s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
-        PulsarFailoverDelayMsOptionKey -> "0"))
+      PulsarFailoverConfig.fromParams(
+        Map(
+          PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+          s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+          PulsarFailoverDelayMsOptionKey -> "0"))
     }
 
-    assert(error.getMessage.contains(PulsarFailoverDelayMsOptionKey))
+    assert(error.getMessage.contains(PulsarFailoverDelayMsDisplayKey))
     assert(error.getMessage.contains("positive"))
   }
 
   test("full config parses durations, policy, auth and tls per secondary") {
-    val config = PulsarFailoverConfig.fromParams(Map(
-      PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
-      PulsarFailoverDelayMsOptionKey -> "5000",
-      PulsarFailoverSwitchBackDelayMsOptionKey -> "10000",
-      PulsarFailoverCheckIntervalMsOptionKey -> "15000",
-      PulsarFailoverPolicyOptionKey -> "order",
-      s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
-      s"${PulsarFailoverSecondaryPrefix}0.authPluginClassName" -> "plugin0",
-      s"${PulsarFailoverSecondaryPrefix}0.authParams" -> "params0",
-      s"${PulsarFailoverSecondaryPrefix}0.tlsTrustCertsFilePath" -> "/cert0.pem",
-      s"${PulsarFailoverSecondaryPrefix}0.tlsTrustStorePath" -> "/truststore0.jks",
-      s"${PulsarFailoverSecondaryPrefix}0.tlsTrustStorePassword" -> "password0",
-      s"${PulsarFailoverSecondaryPrefix}1.serviceUrl" -> secondary1Url,
-      s"${PulsarFailoverSecondaryPrefix}1.authPluginClassName" -> "plugin1",
-      s"${PulsarFailoverSecondaryPrefix}1.authParams" -> "params1",
-      s"${PulsarFailoverSecondaryPrefix}1.tlsTrustCertsFilePath" -> "/cert1.pem",
-      s"${PulsarFailoverSecondaryPrefix}1.tlsTrustStorePath" -> "/truststore1.jks",
-      s"${PulsarFailoverSecondaryPrefix}1.tlsTrustStorePassword" -> "password1")).get
+    val config = PulsarFailoverConfig
+      .fromParams(
+        Map(
+          PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+          PulsarFailoverDelayMsOptionKey -> "5000",
+          PulsarFailoverSwitchBackDelayMsOptionKey -> "10000",
+          PulsarFailoverCheckIntervalMsOptionKey -> "15000",
+          PulsarFailoverPolicyOptionKey -> "order",
+          s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+          s"${PulsarFailoverSecondaryPrefix}0.authPluginClassName" -> "plugin0",
+          s"${PulsarFailoverSecondaryPrefix}0.authParams" -> "params0",
+          s"${PulsarFailoverSecondaryPrefix}0.tlsTrustCertsFilePath" -> "/cert0.pem",
+          s"${PulsarFailoverSecondaryPrefix}0.tlsTrustStorePath" -> "/truststore0.jks",
+          s"${PulsarFailoverSecondaryPrefix}0.tlsTrustStorePassword" -> "password0",
+          s"${PulsarFailoverSecondaryPrefix}1.serviceUrl" -> secondary1Url,
+          s"${PulsarFailoverSecondaryPrefix}1.authPluginClassName" -> "plugin1",
+          s"${PulsarFailoverSecondaryPrefix}1.authParams" -> "params1",
+          s"${PulsarFailoverSecondaryPrefix}1.tlsTrustCertsFilePath" -> "/cert1.pem",
+          s"${PulsarFailoverSecondaryPrefix}1.tlsTrustStorePath" -> "/truststore1.jks",
+          s"${PulsarFailoverSecondaryPrefix}1.tlsTrustStorePassword" -> "password1"))
+      .get
 
     assert(config.primaryServiceUrl === primaryUrl)
     assert(config.failoverDelay === Duration.ofMillis(5000))
     assert(config.switchBackDelay === Duration.ofMillis(10000))
     assert(config.checkInterval === Duration.ofMillis(15000))
     assert(config.policy === FailoverPolicy.ORDER)
-    assert(config.secondaries === Seq(
-      SecondaryClusterConfig(
-        secondary0Url,
-        Some("plugin0" -> "params0"),
-        Some("/cert0.pem"),
-        Some("/truststore0.jks"),
-        Some("password0")),
-      SecondaryClusterConfig(
-        secondary1Url,
-        Some("plugin1" -> "params1"),
-        Some("/cert1.pem"),
-        Some("/truststore1.jks"),
-        Some("password1"))))
+    assert(
+      config.secondaries === Seq(
+        SecondaryClusterConfig(
+          secondary0Url,
+          Some("plugin0" -> "params0"),
+          Some("/cert0.pem"),
+          Some("/truststore0.jks"),
+          Some("password0")),
+        SecondaryClusterConfig(
+          secondary1Url,
+          Some("plugin1" -> "params1"),
+          Some("/cert1.pem"),
+          Some("/truststore1.jks"),
+          Some("password1"))))
   }
 
   test("secondary auth plugin and params must be configured together") {
     val error = intercept[IllegalArgumentException] {
-      PulsarFailoverConfig.fromParams(Map(
-        PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
-        s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
-        s"${PulsarFailoverSecondaryPrefix}0.authPluginClassName" -> "plugin0"))
+      PulsarFailoverConfig.fromParams(
+        Map(
+          PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+          s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+          s"${PulsarFailoverSecondaryPrefix}0.authPluginClassName" -> "plugin0"))
     }
 
     assert(error.getMessage.contains("authPluginClassName"))
     assert(error.getMessage.contains("authParams"))
   }
 
+  test("stray failover options do not enable failover without primary service url") {
+    assert(
+      PulsarFailoverConfig
+        .fromParams(
+          Map(ServiceUrlOptionKey -> primaryUrl, PulsarFailoverDelayMsOptionKey -> "5000"))
+        .isEmpty)
+  }
+
   test("toServiceUrlProvider builds AutoClusterFailover provider") {
     val config = PulsarFailoverConfig(
       primaryUrl,
@@ -126,7 +143,12 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
 
     val provider = PulsarFailoverConfig.toServiceUrlProvider(config)
 
-    assert(provider.getClass.getName.contains("AutoClusterFailover"))
-    assert(provider.getServiceUrl === primaryUrl)
+    val failover = provider.asInstanceOf[org.apache.pulsar.client.impl.AutoClusterFailover]
+    assert(failover.getPrimary === primaryUrl)
+    assert(failover.getSecondary === Seq(secondary0Url).asJava)
+    assert(failover.getFailoverPolicy === FailoverPolicy.ORDER)
+    assert(failover.getFailoverDelayNs === Duration.ofMillis(5000).toNanos)
+    assert(failover.getSwitchBackDelayNs === Duration.ofMillis(10000).toNanos)
+    assert(failover.getIntervalMs === 15000L)
   }
 }

From b70f5da6e1c62f257933d60af42ac206351b2248 Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Thu, 14 May 2026 16:11:55 +0800
Subject: [PATCH 3/7] Strengthen Pulsar failover config tests

---
 .../pulsar/PulsarFailoverConfigSuite.scala    | 42 ++++++++++++++++++-
 1 file changed, 40 insertions(+), 2 deletions(-)

diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
index 1f8fb4f..9961f09 100644
--- a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
@@ -67,6 +67,25 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
     assert(error.getMessage.contains("positive"))
   }
 
+  test("documented option keys parse case-insensitively") {
+    val config = PulsarFailoverConfig
+      .fromParams(
+        Map(
+          PulsarFailoverPrimaryServiceUrlDisplayKey.toUpperCase(java.util.Locale.ROOT) ->
+            primaryUrl,
+          PulsarFailoverDelayMsDisplayKey -> "5000",
+          PulsarFailoverSwitchBackDelayMsDisplayKey -> "10000",
+          PulsarFailoverCheckIntervalMsDisplayKey -> "15000",
+          s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url))
+      .get
+
+    assert(config.primaryServiceUrl === primaryUrl)
+    assert(config.failoverDelay === Duration.ofMillis(5000))
+    assert(config.switchBackDelay === Duration.ofMillis(10000))
+    assert(config.checkInterval === Duration.ofMillis(15000))
+    assert(config.secondaries.map(_.serviceUrl) === Seq(secondary0Url))
+  }
+
   test("full config parses durations, policy, auth and tls per secondary") {
     val config = PulsarFailoverConfig
       .fromParams(
@@ -135,7 +154,14 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
   test("toServiceUrlProvider builds AutoClusterFailover provider") {
     val config = PulsarFailoverConfig(
       primaryUrl,
-      Seq(SecondaryClusterConfig(secondary0Url, None, None, None, None)),
+      Seq(
+        SecondaryClusterConfig(
+          secondary0Url,
+          Some("org.apache.pulsar.client.impl.auth.AuthenticationToken" -> "token:secondary"),
+          Some("/cert0.pem"),
+          Some("/truststore0.jks"),
+          Some("password0")),
+        SecondaryClusterConfig(secondary1Url, None, None, None, None)),
       Duration.ofMillis(5000),
       Duration.ofMillis(10000),
       Duration.ofMillis(15000),
@@ -145,10 +171,22 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
 
     val failover = provider.asInstanceOf[org.apache.pulsar.client.impl.AutoClusterFailover]
     assert(failover.getPrimary === primaryUrl)
-    assert(failover.getSecondary === Seq(secondary0Url).asJava)
+    assert(failover.getSecondary === Seq(secondary0Url, secondary1Url).asJava)
     assert(failover.getFailoverPolicy === FailoverPolicy.ORDER)
     assert(failover.getFailoverDelayNs === Duration.ofMillis(5000).toNanos)
     assert(failover.getSwitchBackDelayNs === Duration.ofMillis(10000).toNanos)
     assert(failover.getIntervalMs === 15000L)
+    assert(failover.getSecondaryAuthentications.size() === 2)
+    assert(failover.getSecondaryAuthentications.get(secondary0Url) !== null)
+    assert(failover.getSecondaryAuthentications.get(secondary1Url) === null)
+    assert(
+      failover.getSecondaryTlsTrustCertsFilePaths.asScala.toMap ===
+        Map(secondary0Url -> "/cert0.pem", secondary1Url -> null))
+    assert(
+      failover.getSecondaryTlsTrustStorePaths.asScala.toMap ===
+        Map(secondary0Url -> "/truststore0.jks", secondary1Url -> null))
+    assert(
+      failover.getSecondaryTlsTrustStorePasswords.asScala.toMap ===
+        Map(secondary0Url -> "password0", secondary1Url -> null))
   }
 }

From 4a0be0f9c9b5f9f715d56378b471723011dc1f07 Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Thu, 14 May 2026 16:55:36 +0800
Subject: [PATCH 4/7] Add Pulsar AutoClusterFailover support tests

---
 .../sql/pulsar/PulsarFailoverConfig.scala     |  2 +-
 .../PulsarFailoverIntegrationSuite.scala      | 38 +++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)
 create mode 100644 src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala

diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
index 1e5cf24..9965783 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
@@ -13,10 +13,10 @@
  */
 package org.apache.spark.sql.pulsar
 
+import java.{util => ju}
 import java.time.Duration
 import java.util.Locale
 import java.util.concurrent.TimeUnit
-import java.{util => ju}
 
 import scala.jdk.CollectionConverters._
 
diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala
new file mode 100644
index 0000000..fc09dd2
--- /dev/null
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala
@@ -0,0 +1,38 @@
+/*
+ * Licensed 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.spark.sql.pulsar
+
+class PulsarFailoverIntegrationSuite extends PulsarSourceTest {
+  import PulsarOptions._
+  import testImplicits._
+
+  test("read from primary with AutoClusterFailover when service.url is omitted") {
+    val topic = newTopic()
+    sendMessages(topic, (1 to 3).map(_.toString).toArray)
+
+    val pulsar = spark.readStream
+      .format("pulsar")
+      .option(PulsarFailoverPrimaryServiceUrlDisplayKey, serviceUrl)
+      .option(s"${PulsarFailoverSecondaryPrefix}0.serviceUrl", "pulsar://127.0.0.1:1")
+      .option(StartingOffsetsOptionKey, "earliest")
+      .option(TopicSingle, topic)
+      .load()
+      .selectExpr("CAST(value AS STRING)")
+      .as[String]
+
+    testStream(pulsar)(
+      makeSureGetOffsetCalled,
+      CheckAnswer("1", "2", "3"))
+  }
+}

From 58929baf86390eb0343a03a624c7f8d8181281ae Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Thu, 14 May 2026 19:21:31 +0800
Subject: [PATCH 5/7] Format Pulsar failover integration test

---
 .../spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala     | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala
index fc09dd2..0a36a16 100644
--- a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala
@@ -31,8 +31,6 @@ class PulsarFailoverIntegrationSuite extends PulsarSourceTest {
       .selectExpr("CAST(value AS STRING)")
       .as[String]
 
-    testStream(pulsar)(
-      makeSureGetOffsetCalled,
-      CheckAnswer("1", "2", "3"))
+    testStream(pulsar)(makeSureGetOffsetCalled, CheckAnswer("1", "2", "3"))
   }
 }

From 0538bd64408d8984fc44eca528dc4709731d4abb Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Fri, 15 May 2026 22:47:03 +0800
Subject: [PATCH 6/7] address failover review comments

---
 README.md                                     |  2 +-
 .../spark/sql/pulsar/PulsarProvider.scala     | 78 ++++++++++---------
 .../pulsar/PulsarFailoverConfigSuite.scala    | 19 +++++
 3 files changed, 60 insertions(+), 39 deletions(-)

diff --git a/README.md b/README.md
index af16464..3940f60 100644
--- a/README.md
+++ b/README.md
@@ -533,7 +533,7 @@ Primary cluster authentication and TLS use existing `pulsar.client.*` options. S
 .option("pulsar.failover.secondary.0.tlsTrustStorePassword", "<password>")
 ```
 
-Secondary indexes must start at `0` and be continuous. At least one secondary cluster is required; use plain `service.url` if no secondary cluster is needed. Only `ORDER` policy is supported now.
+Secondary indexes must start at `0` and be continuous. At least one secondary cluster is required; use plain `service.url` if no secondary cluster is needed. The connector accepts Pulsar client `AutoClusterFailover` policies through `pulsar.failover.policy`; with the current Pulsar client, `ORDER` is the available policy and default.
 
 Limitations:
 
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
index 5d69ac2..b82fe85 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarProvider.scala
@@ -395,7 +395,34 @@ private[pulsar] object PulsarProvider extends Logging {
   private def getServiceUrl(parameters: Map[String, String]): String = {
     PulsarFailoverConfig
       .primaryServiceUrl(parameters)
-      .getOrElse(parameters(ServiceUrlOptionKey))
+      .orElse(parameters.get(ServiceUrlOptionKey).map(_.trim))
+      .get
+  }
+
+  private def validateServiceUrlOptions(caseInsensitiveParams: Map[String, String]): Unit = {
+    val failoverConfig = PulsarFailoverConfig.fromParams(caseInsensitiveParams)
+    failoverConfig match {
+      case Some(config) =>
+        caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
+          require(
+            serviceUrl.trim == config.primaryServiceUrl,
+            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlDisplayKey " +
+              "when Pulsar failover is enabled")
+        }
+      case None =>
+        require(
+          caseInsensitiveParams.get(ServiceUrlOptionKey).exists(_.trim.nonEmpty),
+          s"$ServiceUrlOptionKey must be specified")
+    }
+  }
+
+  private def withUnloggedParams(
+      loggedParams: ju.Map[String, Object],
+      unloggedParams: Map[String, String]): ju.Map[String, Object] = {
+    unloggedParams.foreach { case (key, value) =>
+      loggedParams.put(key, value)
+    }
+    loggedParams
   }
 
   private def getAdminUrl(parameters: Map[String, String]): Option[String] = {
@@ -423,20 +450,7 @@ private[pulsar] object PulsarProvider extends Logging {
 
   private def validateGeneralOptions(
       caseInsensitiveParams: Map[String, String]): Map[String, String] = {
-    val failoverConfig = PulsarFailoverConfig.fromParams(caseInsensitiveParams)
-    failoverConfig match {
-      case Some(config) =>
-        caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
-          require(
-            serviceUrl == config.primaryServiceUrl,
-            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlDisplayKey " +
-              "when Pulsar failover is enabled")
-        }
-      case None =>
-        require(
-          caseInsensitiveParams.contains(ServiceUrlOptionKey),
-          s"$ServiceUrlOptionKey must be specified")
-    }
+    validateServiceUrlOptions(caseInsensitiveParams)
 
     // validate topic options
     val topicOptions = caseInsensitiveParams.filter { case (k, _) =>
@@ -521,20 +535,7 @@ private[pulsar] object PulsarProvider extends Logging {
   private def validateSinkOptions(parameters: Map[String, String]): Map[String, String] = {
     val caseInsensitiveParams = parameters.map { case (k, v) => (k.toLowerCase(Locale.ROOT), v) }
 
-    val failoverConfig = PulsarFailoverConfig.fromParams(caseInsensitiveParams)
-    failoverConfig match {
-      case Some(config) =>
-        caseInsensitiveParams.get(ServiceUrlOptionKey).foreach { serviceUrl =>
-          require(
-            serviceUrl == config.primaryServiceUrl,
-            s"$ServiceUrlOptionKey must match $PulsarFailoverPrimaryServiceUrlDisplayKey " +
-              "when Pulsar failover is enabled")
-        }
-      case None =>
-        require(
-          caseInsensitiveParams.contains(ServiceUrlOptionKey),
-          s"$ServiceUrlOptionKey must be specified")
-    }
+    validateServiceUrlOptions(caseInsensitiveParams)
 
     val topicOptions =
       caseInsensitiveParams.filter { case (k, _) => TopicOptionKeys.contains(k) }.toSeq.toMap
@@ -557,13 +558,15 @@ private[pulsar] object PulsarProvider extends Logging {
 
     val serviceUrl = getServiceUrl(parameters)
     val adminUrl = getAdminUrl(parameters)
-    var clientParams = getClientParams(parameters) ++ getFailoverParams(parameters)
-    clientParams += (ServiceUrlOptionKey -> serviceUrl)
+    val clientParams = getClientParams(parameters) + (ServiceUrlOptionKey -> serviceUrl)
     val readerParams = getReaderParams(parameters)
     val adminParams = getAdminParams(parameters)
 
+    val clientConfig = paramsToPulsarConf("pulsar.client", clientParams)
+    withUnloggedParams(clientConfig, getFailoverParams(parameters))
+
     (
-      paramsToPulsarConf("pulsar.client", clientParams),
+      clientConfig,
       paramsToPulsarConf("pulsar.reader", readerParams),
       paramsToPulsarConf("pulsar.admin", adminParams),
       serviceUrl,
@@ -575,16 +578,15 @@ private[pulsar] object PulsarProvider extends Logging {
 
     val serviceUrl = getServiceUrl(parameters)
 
-    var clientParams = getClientParams(parameters) ++ getFailoverParams(parameters)
-    clientParams += (ServiceUrlOptionKey -> serviceUrl)
+    val clientParams = getClientParams(parameters) + (ServiceUrlOptionKey -> serviceUrl)
     val producerParams = getProducerParams(parameters)
 
     val topic = parameters.get(TopicSingle).map(_.trim).map(TopicName.get(_).toString)
 
-    (
-      paramsToPulsarConf("pulsar.client", clientParams),
-      paramsToPulsarConf("pulsar.producer", producerParams),
-      topic)
+    val clientConfig = paramsToPulsarConf("pulsar.client", clientParams)
+    withUnloggedParams(clientConfig, getFailoverParams(parameters))
+
+    (clientConfig, paramsToPulsarConf("pulsar.producer", producerParams), topic)
   }
 
   private def jsonOptions: JSONOptionsInRead = {
diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
index 9961f09..b47dc05 100644
--- a/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
@@ -32,6 +32,13 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
     assert(PulsarFailoverConfig.fromParams(Map(ServiceUrlOptionKey -> primaryUrl)).isEmpty)
   }
 
+  test("primary service URL is trimmed") {
+    assert(
+      PulsarFailoverConfig
+        .primaryServiceUrl(Map(PulsarFailoverPrimaryServiceUrlOptionKey -> s" $primaryUrl "))
+        .contains(primaryUrl))
+  }
+
   test("missing secondary.0.serviceUrl fails with guidance to use service.url") {
     val error = intercept[IllegalArgumentException] {
       PulsarFailoverConfig.fromParams(Map(PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl))
@@ -151,6 +158,18 @@ class PulsarFailoverConfigSuite extends SparkFunSuite {
         .isEmpty)
   }
 
+  test("unsupported policy fails") {
+    val error = intercept[IllegalArgumentException] {
+      PulsarFailoverConfig.fromParams(
+        Map(
+          PulsarFailoverPrimaryServiceUrlOptionKey -> primaryUrl,
+          s"${PulsarFailoverSecondaryPrefix}0.serviceUrl" -> secondary0Url,
+          PulsarFailoverPolicyOptionKey -> "round_robin"))
+    }
+
+    assert(error.getMessage.contains("Unsupported Pulsar failover policy"))
+  }
+
   test("toServiceUrlProvider builds AutoClusterFailover provider") {
     val config = PulsarFailoverConfig(
       primaryUrl,

From c127bfac7086ebb688d1e25a7809622de6b039ad Mon Sep 17 00:00:00 2001
From: Rui Fu <rfu@apache.org>
Date: Tue, 26 May 2026 16:40:01 +0800
Subject: [PATCH 7/7] test(pulsar): ensure resource cleanup in PulsarTest

---
 .../apache/spark/sql/pulsar/PulsarTest.scala  | 41 +++++++++++--------
 1 file changed, 25 insertions(+), 16 deletions(-)

diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarTest.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarTest.scala
index 03c8aee..4ca64d7 100644
--- a/src/test/scala/org/apache/spark/sql/pulsar/PulsarTest.scala
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarTest.scala
@@ -43,7 +43,7 @@ import org.apache.spark.util.Utils
 import java.util.concurrent.atomic.AtomicInteger
 
 /**
- * A trait to clean cached Pulsar producers in `afterAll`
+ * A trait to clean cached Pulsar resources in `afterAll`
  */
 trait PulsarTest extends BeforeAndAfterAll with BeforeAndAfterEach {
   self: SparkFunSuite =>
@@ -80,12 +80,16 @@ trait PulsarTest extends BeforeAndAfterAll with BeforeAndAfterEach {
     s"subscription-${subscriptionId.getAndIncrement()}").toString
 
   override def afterAll(): Unit = {
-    super.afterAll()
+    CachedConsumer.clear()
     CachedPulsarClient.clear()
-    if (pulsarContainer != null) {
-      pulsarContainer.stop()
-      pulsarContainer.close()
-      brokerConfigs.clear()
+    try {
+      super.afterAll()
+    } finally {
+      if (pulsarContainer != null) {
+        pulsarContainer.stop()
+        pulsarContainer.close()
+        brokerConfigs.clear()
+      }
     }
   }
 
@@ -282,17 +286,22 @@ trait PulsarTest extends BeforeAndAfterAll with BeforeAndAfterEach {
       .serviceUrl(serviceUrl)
       .build()
 
-    val topicPartitions = topics.flatMap { tp =>
-      client.getPartitionsForTopic(tp).get().asScala
+    try {
+      val topicPartitions = topics.flatMap { tp =>
+        client.getPartitionsForTopic(tp).get().asScala
+      }
+      val subscription = newSubscription()
+      try {
+        topicPartitions.map { tp =>
+          val mid = CachedConsumer.getOrCreate(tp, subscription, client).getLastMessageId
+          tp -> mid
+        }.toMap
+      } finally {
+        topicPartitions.foreach(CachedConsumer.close(_, subscription))
+      }
+    } finally {
+      client.close()
     }
-    val subscription = newSubscription()
-    val offsets = topicPartitions.map { tp =>
-      val mid = CachedConsumer.getOrCreate(tp, subscription, client).getLastMessageId
-      tp -> mid
-    }.toMap
-    client.close()
-    topicPartitions.foreach(CachedConsumer.close(_, subscription))
-    offsets
   }
 
   def addPartitions(topic: String, partitions: Int): Unit = {