You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "holdenk (via GitHub)" <gi...@apache.org> on 2026/09/08 13:09:33 UTC
[PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
holdenk opened a new pull request, #58620:
URL: https://github.com/apache/spark/pull/58620
### What changes were proposed in this pull request?
We should validate data when we are recovering from driver failure in a high availability deployment.
### Why are the changes needed?
If we're recovering from failure it's possible we've got invalid data as well.
### Does this PR introduce _any_ user-facing change?
Configurable filtering on classes during ZK recovery.
### How was this patch tested?
New unit test
### Was this patch authored or co-authored using generative AI tooling?
Yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4042469736
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
Review Comment:
I have a 3.5 branch version which uses JDK8 primitives fro filtering but yes happy to drop the backport text and we'll just have the version show up :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4085349519
##########
core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala:
##########
@@ -84,6 +84,9 @@ private[master] class FileSystemPersistenceEngine(
}
}
+ // Unlike ZooKeeperPersistenceEngine, no recovery serialization filter is applied here:
+ // the store is local to the master host; if it is corrupted, the master cannot trust
Review Comment:
nit: "the store is local to the master host" isn't always true. `docs/spark-standalone.md` explicitly describes mounting an NFS directory as the recovery directory and restarting the Master on a different node. Maybe word it as a lower-exposure surface rather than a local-only one.
##########
core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala:
##########
@@ -68,13 +75,48 @@ private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serializer
private def deserializeFromFile[T](filename: String)(implicit m: ClassTag[T]): Option[T] = {
val fileData = zk.getData().forPath(workingDir + "/" + filename)
+ val recordingFilter = new RecordingFilter(serializationFilter)
try {
- Some(serializer.newInstance().deserialize[T](ByteBuffer.wrap(fileData)))
+ serializer.newInstance() match {
+ case javaInstance: JavaSerializerInstance =>
+ val in = javaInstance.deserializeStream(
+ new ByteBufferInputStream(ByteBuffer.wrap(fileData)), recordingFilter)
+ try {
+ Some(in.readObject[T]())
+ } finally {
+ in.close()
+ }
+ case instance =>
+ Some(instance.deserialize[T](ByteBuffer.wrap(fileData)))
+ }
} catch {
+ case e: Exception if recordingFilter.rejected =>
+ // Rejected by the serialization filter, not found corrupt. Skip the znode without
+ // deleting it: an overly narrow filter pattern (e.g. "org.apache.spark.*", which
+ // does not match subpackages) must not wipe the whole recovery state on failover.
+ logError(s"Skipping persisted file $filename, rejected by the recovery " +
Review Comment:
nit: Could you use structured logging here, e.g. `log"Skipping persisted file ${MDC(LogKeys.FILE_NAME, filename)}, rejected by the recovery serialization filter (${MDC(LogKeys.CONFIG, RECOVERY_SERIALIZATION_FILTER.key)})"`?
##########
docs/spark-standalone.md:
##########
@@ -857,6 +857,22 @@ In order to enable this recovery mode, you can set SPARK_DAEMON_JAVA_OPTS in spa
</td>
<td>0.8.1</td>
</tr>
+ <tr>
+ <td><code>spark.deploy.recoverySerializationFilter</code></td>
Review Comment:
nit: Since skipped znodes are never cleaned up by the Master (and are logged on every failover), could you mention that operators should inspect and remove them manually? Also, since this is ZOOKEEPER-only, placing the row next to `spark.deploy.zookeeper.dir` would read more naturally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r3974115487
##########
core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala:
##########
@@ -39,6 +41,12 @@ private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serializer
private val workingDir = conf.get(ZOOKEEPER_DIRECTORY).getOrElse("/spark") + "/master_status"
private val zk: CuratorFramework = SparkCuratorUtil.newClient(conf)
+ // Only instantiate well-known classes while reading persisted state back, so corrupted
+ // or unexpected znode contents are dropped instead of being instantiated in the newly
+ // elected master.
+ private val serializationFilter: ObjectInputFilter =
+ ObjectInputFilter.Config.createFilter(conf.get(RECOVERY_SERIALIZATION_FILTER))
Review Comment:
`spark.deploy.recoverySerializationFilter` is read only here. `FileSystemPersistenceEngine` and `RocksDBPersistenceEngine` deserialize the same `ApplicationInfo`/`DriverInfo`/`WorkerInfo` state through the same unfiltered `JavaSerializer`, so the config's recovery-wide name and doc don't actually cover them. Is limiting it to ZK intentional for now - the file/RocksDB stores are local to the master host, a lower-exposure surface than a shared ZooKeeper ensemble - with the other engines as follow-up? If so a one-line note would help; if not, the same `JavaDeserializationStream` filter applies directly in their deserialize paths.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r3985584697
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
+ "all versions after 4.3.0.")
+ .version("4.3.0")
Review Comment:
This should be `4.4.0`. `v4.3.0-rc1` was already tagged on 2026-08-31, before this PR was opened, and `branch-4.3` is now at `4.3.1-SNAPSHOT` while `branch-4.x` is at `4.4.0-SNAPSHOT`. The same applies to `<td>4.3.0</td>` in `docs/spark-standalone.md`.
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
Review Comment:
Wording (here, in the docs, and in the PR description): the filter blocks deserialization gadgets, but it does not make ZooKeeper contents trustworthy, so "validate data" / "treated as corrupt" overstate it. Allowlisted Spark classes can still carry arbitrary data from anyone with ZooKeeper write access; e.g. `Master.completeRecovery` relaunches unclaimed supervised drivers using the persisted `DriverDescription.command`. Also, genuinely corrupted bytes were already dropped via `StreamCorruptedException` before this PR.
Maybe describe it as deserialization hardening, and note that ZooKeeper ACLs remain the actual access control?
##########
core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala:
##########
@@ -69,7 +77,21 @@ private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serializer
private def deserializeFromFile[T](filename: String)(implicit m: ClassTag[T]): Option[T] = {
val fileData = zk.getData().forPath(workingDir + "/" + filename)
try {
- Some(serializer.newInstance().deserialize[T](ByteBuffer.wrap(fileData)))
+ val inputStream = new ByteBufferInputStream(ByteBuffer.wrap(fileData))
+ val in = serializer match {
+ case _: JavaSerializer =>
+ // A class rejected by the filter surfaces as InvalidClassException, which lands
+ // in the catch below: the znode is deleted and recovery continues.
+ new JavaDeserializationStream(
+ inputStream, Utils.getContextOrSparkClassLoader, Some(serializationFilter))
+ case _ =>
+ serializer.newInstance().deserializeStream(inputStream)
+ }
+ try {
+ Some(in.readObject[T]())
+ } finally {
+ in.close()
+ }
} catch {
case e: Exception =>
logWarning("Exception while reading persisted file, deleting", e)
Review Comment:
With the filter in place, a pattern that is syntactically valid but slightly too narrow irreversibly deletes the whole recovery state on failover. For example, `org.apache.spark.*` (single `*`) matches only that package, not its subpackages, so every `ApplicationInfo`/`DriverInfo`/`WorkerInfo` znode would be rejected and deleted here.
I verified the semantics on JDK 17: `java.lang.*;java.util.*;!*` rejects `java.util.concurrent.atomic.AtomicInteger` with `InvalidClassException: filter status: REJECTED`, while `java.lang.**;java.util.**;!*` accepts it.
The delete-on-failure behavior is pre-existing, but the filter makes an operator typo much more destructive. How about logging an error and skipping, instead of deleting, when the rejection comes from the filter?
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
Review Comment:
Could you remove the `Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; ...` sentence here and in `docs/spark-standalone.md`? No other config in `internal/config` documents backport versions like this, and it commits to backports that have not been decided yet.
Also, `3.5.10` is not possible as written: `branch-3.5` still builds with `java.version` 1.8, and `java.io.ObjectInputFilter` does not exist on JDK 8 (only `sun.misc.ObjectInputFilter`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r3985584697
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
+ "all versions after 4.3.0.")
+ .version("4.3.0")
Review Comment:
This should be `4.4.0`. `v4.3.0-rc1` was already tagged on 2026-08-31, before this PR was opened, and `branch-4.3` is now at `4.3.1-SNAPSHOT` while `branch-4.x` is at `4.4.0-SNAPSHOT`. The same applies to `<td>4.3.0</td>` in `docs/spark-standalone.md`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Apply a serialization filter when recovering Master state from ZooKeeper [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #58620: [SPARK-59333][CORE] Apply a serialization filter when recovering Master state from ZooKeeper
URL: https://github.com/apache/spark/pull/58620
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4049203412
##########
core/src/main/scala/org/apache/spark/serializer/JavaSerializer.scala:
##########
@@ -85,6 +88,17 @@ private[spark] class JavaDeserializationStream(in: InputStream, loader: ClassLoa
}
+ // A JEP-290 deserialization filter for callers that validate persisted data on read
+ // (e.g. the master recovery store). Applied per-stream so it cannot affect other
+ // JavaSerializer users. If the stream already has a filter (e.g. a JVM-wide
+ // jdk.serialFilter), compose the two instead of replacing it.
+ filter.foreach { newFilter =>
Review Comment:
`composeFilters` has the same semantics as JDK 17's `ObjectInputFilter.merge(filter, anotherFilter)` (REJECTED if either rejects, else ALLOWED if either allows, else UNDECIDED). Since Spark requires Java 17, could we use it directly? Then `JavaDeserializationStream` object doesn't need to be widened to `private[spark]`, and the new `JavaSerializerSuite` test is no longer needed.
```scala
filter.foreach { f =>
objIn.setObjectInputFilter(
Option(objIn.getObjectInputFilter).map(ObjectInputFilter.merge(_, f)).getOrElse(f))
}
```
##########
core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala:
##########
@@ -69,12 +76,37 @@ private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serializer
private def deserializeFromFile[T](filename: String)(implicit m: ClassTag[T]): Option[T] = {
val fileData = zk.getData().forPath(workingDir + "/" + filename)
try {
- Some(serializer.newInstance().deserialize[T](ByteBuffer.wrap(fileData)))
+ val inputStream = new ByteBufferInputStream(ByteBuffer.wrap(fileData))
+ val in = serializer match {
+ case _: JavaSerializer =>
+ // A class rejected by the filter surfaces as InvalidClassException.
+ new JavaDeserializationStream(
+ inputStream, Utils.getContextOrSparkClassLoader, Some(serializationFilter))
+ case _ =>
Review Comment:
Master always passes a `JavaSerializer` (`Master.scala`), so this branch is effectively test-only. How about keeping the original `serializer.newInstance().deserialize[T](ByteBuffer.wrap(fileData))` here to minimize the change?
Also, the `JavaSerializer` branch above uses `Utils.getContextOrSparkClassLoader` directly, which bypasses `JavaSerializer.setDefaultClassLoader`. Adding a filter-aware `deserializeStream` overload to `JavaSerializerInstance` might be more natural.
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,19 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master reads back " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields). Znodes containing any other class " +
+ "are skipped during recovery instead of being instantiated in the newly " +
+ "elected master. Set to '*' to disable filtering.")
+ .version("4.3.0")
+ .stringConf
+ .createWithDefault("java.**;scala.**;org.apache.spark.**;!*")
Review Comment:
An empty value behaves inconsistently because `ObjectInputFilter.Config.createFilter("")` returns `null`. I verified it on JDK:
- Without `jdk.serialFilter`: `setObjectInputFilter(null)` silently disables the filtering.
- With `jdk.serialFilter`: `composeFilters(existing, null)` throws an NPE in `checkInput`, which the JDK turns into `filter status: REJECTED`, so every znode is skipped during recovery.
Could you add `.checkValue(_.trim.nonEmpty, ...)`?
##########
core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala:
##########
@@ -84,6 +84,9 @@ private[master] class FileSystemPersistenceEngine(
}
}
+ // Unlike ZooKeeperPersistenceEngine, no recovery serialization filter is applied here:
+ // the store is local to the master host; if it is corrupted, the master cannot trust
+ // itself anyways.
Review Comment:
nit: `anyways` -> `anyway` (also in `RocksDBPersistenceEngine`).
##########
core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala:
##########
@@ -132,6 +135,57 @@ class PersistenceEngineSuite extends SparkFunSuite {
}
}
+ test("ZooKeeperPersistenceEngine skips classes outside the serialization filter allowlist") {
Review Comment:
nit: Could you add the JIRA ID prefix, e.g. `test("SPARK-59333: ...")`, like the other tests in this suite?
##########
core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala:
##########
@@ -69,12 +76,37 @@ private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serializer
private def deserializeFromFile[T](filename: String)(implicit m: ClassTag[T]): Option[T] = {
val fileData = zk.getData().forPath(workingDir + "/" + filename)
try {
- Some(serializer.newInstance().deserialize[T](ByteBuffer.wrap(fileData)))
+ val inputStream = new ByteBufferInputStream(ByteBuffer.wrap(fileData))
+ val in = serializer match {
+ case _: JavaSerializer =>
+ // A class rejected by the filter surfaces as InvalidClassException.
+ new JavaDeserializationStream(
+ inputStream, Utils.getContextOrSparkClassLoader, Some(serializationFilter))
+ case _ =>
+ serializer.newInstance().deserializeStream(inputStream)
+ }
+ try {
+ Some(in.readObject[T]())
+ } finally {
+ in.close()
+ }
} catch {
+ case e: InvalidClassException if isFilterRejection(e) =>
+ // Rejected by the serialization filter, not found corrupt. Skip the znode without
+ // deleting it: an overly narrow filter pattern (e.g. "org.apache.spark.*", which
+ // does not match subpackages) must not wipe the whole recovery state on failover.
+ logError(s"Skipping persisted file $filename, rejected by the recovery " +
+ s"serialization filter (${RECOVERY_SERIALIZATION_FILTER.key})", e)
+ None
case e: Exception =>
logWarning("Exception while reading persisted file, deleting", e)
zk.delete().forPath(workingDir + "/" + filename)
None
}
}
+
+ // The JDK reports a filter rejection only as an InvalidClassException whose message is
+ // "filter status: REJECTED"; there is no more specific exception type to match on.
+ private def isFilterRejection(e: InvalidClassException): Boolean =
Review Comment:
nit (optional): matching on the JDK exception message is a bit fragile. Wrapping the filter so that it records when it returns `REJECTED` would avoid depending on the message text.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4042475647
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
Review Comment:
Good point, ZK ACLs are the "real" security here but the filtering is just an extra layer to avoid some accidental mishaps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4078165987
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,19 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master reads back " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields). Znodes containing any other class " +
+ "are skipped during recovery instead of being instantiated in the newly " +
+ "elected master. Set to '*' to disable filtering.")
+ .version("4.3.0")
+ .stringConf
+ .createWithDefault("java.**;scala.**;org.apache.spark.**;!*")
Review Comment:
Did a slightly more involved checkValue :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r3985584702
##########
core/src/main/scala/org/apache/spark/internal/config/Deploy.scala:
##########
@@ -51,6 +51,20 @@ private[spark] object Deploy {
.checkValue(_ > 0, "spark.deploy.recoveryTimeout must be positive.")
.createOptional
+ val RECOVERY_SERIALIZATION_FILTER =
+ ConfigBuilder("spark.deploy.recoverySerializationFilter")
+ .doc("JEP-290 serialization filter pattern applied when the master deserializes " +
+ "recovery state written by the built-in JavaSerializer (currently enforced for " +
+ "the ZOOKEEPER recovery mode). The default allows only JDK, Scala and Spark " +
+ "classes, which covers everything the master persists (ApplicationInfo, " +
+ "DriverInfo, WorkerInfo and their fields); znodes containing anything else are " +
+ "treated as corrupt and dropped during recovery. Set to '*' to disable filtering. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
Review Comment:
Could you remove the `Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; ...` sentence here and in `docs/spark-standalone.md`? No other config in `internal/config` documents backport versions like this, and it commits to backports that have not been decided yet.
Also, `3.5.10` is not possible as written: `branch-3.5` still builds with `java.version` 1.8, and `java.io.ObjectInputFilter` does not exist on JDK 8 (only `sun.misc.ObjectInputFilter`), @holdenk .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4088591953
##########
core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala:
##########
@@ -84,6 +84,9 @@ private[master] class FileSystemPersistenceEngine(
}
}
+ // Unlike ZooKeeperPersistenceEngine, no recovery serialization filter is applied here:
+ // the store is local to the master host; if it is corrupted, the master cannot trust
Review Comment:
Oh true
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r4088590367
##########
docs/spark-standalone.md:
##########
@@ -857,6 +857,22 @@ In order to enable this recovery mode, you can set SPARK_DAEMON_JAVA_OPTS in spa
</td>
<td>0.8.1</td>
</tr>
+ <tr>
+ <td><code>spark.deploy.recoverySerializationFilter</code></td>
Review Comment:
Ah yeah that's a good point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59333][CORE] Validate data during recovery from ZK in HA failure [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58620:
URL: https://github.com/apache/spark/pull/58620#discussion_r3963648363
##########
core/src/main/scala/org/apache/spark/serializer/JavaSerializer.scala:
##########
@@ -85,6 +88,11 @@ private[spark] class JavaDeserializationStream(in: InputStream, loader: ClassLoa
}
+ // A JEP-290 deserialization filter for callers that validate persisted data on read
+ // (e.g. the master recovery store). Applied per-stream so it cannot affect other
+ // JavaSerializer users.
+ filter.foreach(objIn.setObjectInputFilter)
Review Comment:
[P2] Preserve existing JVM deserialization restrictions
When a master already has `jdk.serialFilter` configured, this call replaces the stream's existing JVM filter under the default JDK filter factory. ZooKeeper recovery previously inherited that policy; after this change, existing class restrictions and array/depth limits are silently discarded even with the new Spark setting left at its default. The [JDK contract](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/io/ObjectInputFilter.Config.html#getSerialFilterFactory()) explicitly documents this replacement behavior.
I verified the difference through `ZooKeeperPersistenceEngine`: with `jdk.serialFilter=maxarray=1`, the same harmless two-byte array is rejected on the merge base and recovered on this head. This used the respective changed sources compiled against cached Spark dependencies, rather than a clean full build.
Please compose the recovery filter with `objIn.getObjectInputFilter` so rejection by either policy is preserved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org