Skip to content

[SPARK-59169][CORE] Log a warning when KVStoreProtobufSerializer falls back to JSON SerDe #58467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pan3793 wants to merge 6 commits into
base: master
Choose a base branch
from
+55 −3
Open

[SPARK-59169][CORE] Log a warning when KVStoreProtobufSerializer falls back to JSON SerDe #58467

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a12f79c
[SPARK-59169][CORE] Log a warning when KVStoreProtobufSerializer fall…
pan3793 Sep 2, 2026
6976b68
nit: remove log prefix
pan3793 Sep 2, 2026
7dd4abe
address comment, add UT
pan3793 Sep 2, 2026
cea4510
make UT robust against test ordering
pan3793 Sep 2, 2026
9912008
skip by-design JSON classes, fix wording
pan3793 Sep 2, 2026
2b73e60
keep warning for metadata classes, skip only kvstore internals
pan3793 Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 24 additions & 3 deletions core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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] =
Expand All @@ -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[_] = {
Expand All @@ -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]] = {

@uros-b uros-b Sep 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please take a look at KVStoreProtobufSerializerSuite. No test covers the new warning behavior. The PR description asserts "Existing tests" suffice, but both KVStoreProtobufSerializerSuite suites (core and sql/core) exercise only known-registered classes and never reach the serializer.isEmpty branch. A test using withLogAppender should be added to assert (a) a warning IS logged on the first call with an unregistered class, and (b) is NOT logged again on a repeated call for the same class (the deduplication invariant). Without this, the "warn exactly once" semantics are entirely untested.

@pan3793 pan3793 Sep 2, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a UT in KVStoreProtobufSerializerSuite (core) using withLogAppender, covering both the warning content and the once-per-class dedup, in 7dd4abe.

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)}, " +

@ulysses-you ulysses-you Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think debug level is enough, so nothing noisy by default. We can enhance the display log level by modifying the configuration file.

@pan3793 pan3793 Sep 2, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's intended to log it at the warning level: JSON is the default option, and users likely switch to PROTOBUF for performance purposes, falling back means a performance drop.

for noise concerns, it's a one-time warning that makes the gap visible without flooding the log.

log"falling back to the JSON SerDe.")
}
serializer
}
}
31 changes: 31 additions & 0 deletions core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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") {

@dongjoon-hyun dongjoon-hyun Sep 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: this assertion depends on the process-global missedClasses set never having seen FallbackTestData before. It works today because this test runs first and the class is unique to this suite, but it becomes order-dependent if another test above it (or a re-run in the same JVM) serializes the class first. Maybe worth a short comment, or a private[protobuf] reset hook for tests.

@pan3793 pan3793 Sep 2, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a private[protobuf] reset hook and the test now clears the set up front, so it no longer depends on ordering. Fixed in cea4510.

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",
Expand Down Expand Up @@ -1703,3 +1732,5 @@ class KVStoreProtobufSerializerSuite extends SparkFunSuite {
}
}
}

private[protobuf] case class FallbackTestData(value: String)