diff --git a/README.md b/README.md
index 3937d5f..3940f60 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. The connector accepts Pulsar client `AutoClusterFailover` policies through `pulsar.failover.policy`; with the current Pulsar client, `ORDER` is the available policy and default.
+
+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..e27f2af 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,37 @@ 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, _) =>
+        val normalizedKey = key.toLowerCase(java.util.Locale.ROOT)
+        normalizedKey.startsWith(PulsarFailoverOptionKeyPrefix) ||
+        failoverConfig.isDefined && normalizedKey == ServiceUrlOptionKey
+      }
       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 $PulsarFailoverPrimaryServiceUrlDisplayKey; " +
+                  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..9965783
--- /dev/null
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfig.scala
@@ -0,0 +1,343 @@
+/*
+ * 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.{util => ju}
+import java.time.Duration
+import java.util.Locale
+import java.util.concurrent.TimeUnit
+
+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.collect { case (k, v) if v != null => k -> v.toString })
+  }
+
+  /** 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)
+
+    if (!failoverParams.contains(PulsarFailoverPrimaryServiceUrlOptionKey)) {
+      return None
+    }
+
+    val primary = requiredNonEmpty(
+      failoverParams,
+      PulsarFailoverPrimaryServiceUrlOptionKey,
+      s"$PulsarFailoverPrimaryServiceUrlDisplayKey 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: ${displayKey(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: " +
+          (if (missing.nonEmpty) missing.mkString(",") else "before 0"))
+    }
+
+    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"${displayKey(key)} must be a positive milliseconds value: $value")
+        }
+      case None => defaultMs
+    }
+    if (millis <= 0) {
+      throw new IllegalArgumentException(s"${displayKey(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)
+  }
+
+  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 4b7ac1c..bab9d4c 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,25 @@ 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)
   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..b82fe85 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(
@@ -267,6 +268,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 +393,36 @@ private[pulsar] object PulsarProvider extends Logging {
   }
 
   private def getServiceUrl(parameters: Map[String, String]): String = {
-    parameters(ServiceUrlOptionKey)
+    PulsarFailoverConfig
+      .primaryServiceUrl(parameters)
+      .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] = {
@@ -409,16 +445,12 @@ 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] = {
-    if (!caseInsensitiveParams.contains(ServiceUrlOptionKey)) {
-      throw new IllegalArgumentException(s"$ServiceUrlOptionKey must be specified")
-    }
+    validateServiceUrlOptions(caseInsensitiveParams)
 
     // validate topic options
     val topicOptions = caseInsensitiveParams.filter { case (k, _) =>
@@ -503,9 +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) }
 
-    if (!caseInsensitiveParams.contains(ServiceUrlOptionKey)) {
-      throw new IllegalArgumentException(s"$ServiceUrlOptionKey must be specified")
-    }
+    validateServiceUrlOptions(caseInsensitiveParams)
 
     val topicOptions =
       caseInsensitiveParams.filter { case (k, _) => TopicOptionKeys.contains(k) }.toSeq.toMap
@@ -519,22 +549,28 @@ 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)
-    var clientParams = getClientParams(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, adminUrl)
+      serviceUrl,
+      adminUrl)
   }
 
   private def prepareConfForProducer(parameters: Map[String, String])
@@ -542,16 +578,15 @@ private[pulsar] object PulsarProvider extends Logging {
 
     val serviceUrl = getServiceUrl(parameters)
 
-    var clientParams = getClientParams(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
new file mode 100644
index 0000000..b47dc05
--- /dev/null
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverConfigSuite.scala
@@ -0,0 +1,211 @@
+/*
+ * 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 scala.jdk.CollectionConverters._
+
+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("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))
+    }
+
+    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(PulsarFailoverDelayMsDisplayKey))
+    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(
+        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("stray failover options do not enable failover without primary service url") {
+    assert(
+      PulsarFailoverConfig
+        .fromParams(
+          Map(ServiceUrlOptionKey -> primaryUrl, PulsarFailoverDelayMsOptionKey -> "5000"))
+        .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,
+      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),
+      FailoverPolicy.ORDER)
+
+    val provider = PulsarFailoverConfig.toServiceUrlProvider(config)
+
+    val failover = provider.asInstanceOf[org.apache.pulsar.client.impl.AutoClusterFailover]
+    assert(failover.getPrimary === primaryUrl)
+    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))
+  }
+}
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..0a36a16
--- /dev/null
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarFailoverIntegrationSuite.scala
@@ -0,0 +1,36 @@
+/*
+ * 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"))
+  }
+}
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 = {