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 d87c9e6d59a75..d8bb13e316f6b 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 23cb99bce2ff4..ef983e1b2b6d1 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)