[SPARK-59333][CORE] Validate data during recovery from ZK in HA failure - #58620
[SPARK-59333][CORE] Validate data during recovery from ZK in HA failure#58620holdenk wants to merge 4 commits into
Conversation
Apply a JEP-290 serialization filter (spark.deploy.recoverySerializationFilter, default java.**;scala.**;org.apache.spark.**;!*) when the master reads back recovery state written by the built-in JavaSerializer, so corrupted or unexpected znode contents are dropped instead of being instantiated during recovery. JavaDeserializationStream gains an optional per-stream filter parameter; other JavaSerializer users are unaffected. Co-authored-by: Cursor <cursoragent@cursor.com>
scalastyle bans org.apache.commons.lang3.tuple and points at org.apache.spark.util.Pair, which is no use here: the test needs a class the recovery filter's default allowlist rejects, and org.apache.spark.** is inside that allowlist. MutableInt is serializable, already on core's classpath, and outside java.**/scala.**/org.apache.spark.**. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca>
…rsion A single version tells an operator nothing about a backported config: it does not say which maintenance releases have it. Name the whole set in the doc, and declare the version this branch actually first ships in. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca>
The version note only reached the config doc, so nobody reading the docs tables saw it. Add the row. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca>
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Found one issue with preserving existing JVM deserialization restrictions. The persistence and Java serializer suites passed 12/12 tests in a local partial build; the inline finding was also checked through ZooKeeper recovery against the merge base.
| // 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) |
There was a problem hiding this 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 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.
HyukjinKwon
left a comment
HyukjinKwon
left a comment
There was a problem hiding this comment.
0 blocking, 2 non-blocking, 0 nits.
A focused, well-tested security hardening of ZooKeeper recovery deserialization; one previously-raised correctness concern (composing with an existing JVM-wide filter) is still open, and the config's ZK-only scope is worth a one-line note.
Already raised in existing discussion (1)
- On JDK 17's default filter factory, objIn.setObjectInputFilter(serializationFilter) replaces the stream's process-wide filter instead of composing with it. A master started with a hardened -Djdk.serialFilter (class restrictions or maxarray/maxdepth/maxrefs/maxbytes limits) silently loses those restrictions on the ZK recovery read path, even with spark.deploy.recoverySerializationFilter left at its default. Compose the two - e.g. combine with objIn.getObjectInputFilter() via ObjectInputFilter.merge - so rejection by either policy is preserved. -- existing discussion
Design / architecture (1)
- core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala:48: Question (non-blocking): spark.deploy.recoverySerializationFilter is honored only by the ZooKeeper recovery engine. FileSystemPersistenceEngine and RocksDBPersistenceEngine deserialize the same JavaSerializer-persisted master state (ApplicationInfo/DriverInfo/WorkerInfo) through an unfiltered path, while the config name and doc read as recovery-wide. Is limiting the filter to ZK intentional for this iteration - e.g. because the file/RocksDB stores are local to the master host, a lower-exposure surface than a shared ZooKeeper ensemble - with the other engines tracked as follow-up? If so, a one-line note would help; if not, the same JavaDeserializationStream filter could be applied in those two engines' deserialize paths. -- see inline
Verification
Confirmed the default allowlist covers every serialized field of the persisted info classes (DriverInfo.exception and the other arbitrary-typed fields are @transient, so no third-party class is serialized). Verified that on Spark's minimum JDK 17 the single per-stream setObjectInputFilter call does not throw when a process-wide jdk.serialFilter is set (it replaces it - the basis of the open reviewer finding).
| // 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)) |
There was a problem hiding this 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.
4 participants
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