This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 9de7379b30b6 [SPARK-59169][CORE] Log a warning when
KVStoreProtobufSerializer falls back to JSON SerDe
9de7379b30b6 is described below
commit 9de7379b30b66a5a668d9c7f0f771d19de733d49
Author: Cheng Pan <[email protected]>
AuthorDate: Fri Oct 9 16:45:34 2026 +0800
[SPARK-59169][CORE] Log a warning when KVStoreProtobufSerializer falls back
to JSON SerDe
### What changes were proposed in this pull request?
`KVStoreProtobufSerializer.getSerializer` now logs a warning when no
`ProtobufSerDe` is registered for a class and it silently falls back to the
JSON SerDe. Each missed class is recorded in a concurrent set so the warning is
logged at most once per class, avoiding log flood.
### Why are the changes needed?
The fallback is currently silent, so a missing `ProtobufSerDe` registration
(e.g. a new UI store class without a corresponding serializer) is easy to miss
and quietly loses the performance benefit of Protobuf serialization. A one-time
warning makes the gap visible without flooding the log.
### Does this PR introduce _any_ user-facing change?
No, only a new warning log message.
### How was this patch tested?
New UT in `KVStoreProtobufSerializerSuite` (core) verifies the warning
content and the once-per-class dedup via `withLogAppender`. Existing
`KVStoreProtobufSerializerSuite` suites in `core` and `sql/core` also pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #58467 from pan3793/SPARK-59169.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
(cherry picked from commit fe185b7f63c8c8f61006bcf4f7396978d8473e9b)
Signed-off-by: Cheng Pan <[email protected]>
---
.../protobuf/KVStoreProtobufSerializer.scala | 27 ++++++++++++++++---
.../protobuf/KVStoreProtobufSerializerSuite.scala | 31 ++++++++++++++++++++++
2 files changed, 55 insertions(+), 3 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
b/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
index d87c9e6d59a7..d8bb13e316f6 100644
---
a/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
+++
b/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
@@ -19,10 +19,14 @@ package org.apache.spark.status.protobuf
import java.lang.reflect.ParameterizedType
import java.util.ServiceLoader
+import java.util.concurrent.ConcurrentHashMap
import scala.jdk.CollectionConverters._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.CLASS_NAME
import org.apache.spark.status.KVUtils.KVStoreScalaSerializer
+import org.apache.spark.util.kvstore.{LevelDB, RocksDB}
private[spark] class KVStoreProtobufSerializer extends KVStoreScalaSerializer {
override def serialize(o: Object): Array[Byte] =
@@ -39,7 +43,7 @@ private[spark] class KVStoreProtobufSerializer extends
KVStoreScalaSerializer {
}
}
-private[spark] object KVStoreProtobufSerializer {
+private[spark] object KVStoreProtobufSerializer extends Logging {
private[this] lazy val serializerMap: Map[Class[_], ProtobufSerDe[Any]] = {
def getGenericsType(klass: Class[_]): Class[_] = {
@@ -51,6 +55,23 @@ private[spark] object KVStoreProtobufSerializer {
}.toMap
}
- def getSerializer(klass: Class[_]): Option[ProtobufSerDe[Any]] =
- serializerMap.get(klass)
+ private[this] val missedClasses = ConcurrentHashMap.newKeySet[Class[_]]()
+
+ // The KVStore backends' own bookkeeping values fall back to the JSON SerDe
by design:
+ // they are internals of the kvstore library, which the ProtobufSerDe SPI
does not cover.
+ // Skip warning for them.
+ private[this] val jsonByDesignClasses: Set[Class[_]] = Set(
+ classOf[LevelDB.TypeAliases],
+ classOf[RocksDB.TypeAliases])
+
+ private[protobuf] def resetMissedClassesForTesting(): Unit =
missedClasses.clear()
+
+ def getSerializer(klass: Class[_]): Option[ProtobufSerDe[Any]] = {
+ val serializer = serializerMap.get(klass)
+ if (serializer.isEmpty && !jsonByDesignClasses.contains(klass) &&
missedClasses.add(klass)) {
+ logWarning(log"No Protobuf SerDe found for class ${MDC(CLASS_NAME,
klass.getName)}, " +
+ log"falling back to the JSON SerDe.")
+ }
+ serializer
+ }
}
diff --git
a/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
b/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
index 23cb99bce2ff..ef983e1b2b6d 100644
---
a/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
+++
b/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
@@ -22,6 +22,8 @@ import java.util.Date
import scala.collection.mutable
import scala.io.Source
+import org.apache.logging.log4j.Level
+
import org.apache.spark.{JobExecutionStatus, SparkFunSuite}
import org.apache.spark.executor.ExecutorMetrics
import org.apache.spark.metrics.ExecutorMetricType
@@ -31,10 +33,37 @@ import org.apache.spark.status._
import org.apache.spark.status.api.v1._
import org.apache.spark.ui.scope.{RDDOperationEdge, RDDOperationNode}
import org.apache.spark.util.Utils.tryWithResource
+import org.apache.spark.util.kvstore.{LevelDB, RocksDB}
class KVStoreProtobufSerializerSuite extends SparkFunSuite {
private val serializer = new KVStoreProtobufSerializer()
+ test("SPARK-59169: log a warning once per class when no ProtobufSerDe is
found") {
+ KVStoreProtobufSerializer.resetMissedClassesForTesting()
+ val appender = new LogAppender("KVStoreProtobufSerializer fallback
warning")
+ withLogAppender(appender, loggerNames =
Seq(classOf[KVStoreProtobufSerializer].getName)) {
+ serializer.serialize(FallbackTestData("a"))
+ serializer.serialize(FallbackTestData("b"))
+ }
+ val warnings = appender.loggingEvents
+ .filter(_.getLevel == Level.WARN)
+ .map(_.getMessage.getFormattedMessage)
+ .filter(_.contains(classOf[FallbackTestData].getName))
+ assert(warnings.size === 1)
+ assert(warnings.head.contains("No Protobuf SerDe found for class"))
+ }
+
+ test("SPARK-59169: no warning for KVStore bookkeeping classes without
ProtobufSerDe") {
+ KVStoreProtobufSerializer.resetMissedClassesForTesting()
+ val appender = new LogAppender("KVStoreProtobufSerializer by-design
fallback")
+ withLogAppender(appender, loggerNames =
Seq(classOf[KVStoreProtobufSerializer].getName)) {
+
assert(KVStoreProtobufSerializer.getSerializer(classOf[RocksDB.TypeAliases]).isEmpty)
+
assert(KVStoreProtobufSerializer.getSerializer(classOf[LevelDB.TypeAliases]).isEmpty)
+ }
+ val warnings = appender.loggingEvents.filter(_.getLevel == Level.WARN)
+ assert(warnings.isEmpty)
+ }
+
test("All the string fields must be optional to avoid NPE") {
val protoFile = getWorkspaceFilePath(
"core", "src", "main", "protobuf", "org", "apache", "spark", "status",
"protobuf",
@@ -1703,3 +1732,5 @@ class KVStoreProtobufSerializerSuite extends
SparkFunSuite {
}
}
}
+
+private[protobuf] case class FallbackTestData(value: String)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
