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: [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]
