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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
