[SPARK-52819][SQL] Making KryoSerializationCodec serializable to fix java.io.NotSerializableException errors in various use cases

### What changes were proposed in this pull request?

This PR makes `KryoSerializationCodec` implements `java.io.Serializable` to avoid `java.io.NotSerializableException` exceptions when using Kryo encoder in `Dataset.flatMapGroupsWithState` or `Aggregator.bufferEncoder`.

### Why are the changes needed?

See the description in [SPARK-52819](https://issues.apache.org/jira/browse/SPARK-52819) as well as the [minimal repro](https://github.com/Kontinuation/spark-4-kryo-encoder-bug). The problems only happens when using Spark 4.0.0 but not when using Spark 3.5.5.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

New tests were added to ensure that
1. Both kryo encoder and java serialization encoder are serializable
2. The kryo encoded aggregator buffer use case works

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#51615 from Kontinuation/fix-kryo-codec-serializable.

Authored-by: Kristin Cowalcijk <bo@wherobots.com>
Signed-off-by: Herman van Hövell <herman@databricks.com>

Author: Kontinuation

sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/codecs.scala

@@ -55,7 +55,7 @@ object JavaSerializationCodec extends (() => Codec[Any, Array[Byte]]) {
  * server (driver & executors) very tricky. As a workaround a user can define their own Codec
  * which internalizes the Kryo configuration.
  */
-object KryoSerializationCodec extends (() => Codec[Any, Array[Byte]]) {
+object KryoSerializationCodec extends (() => Codec[Any, Array[Byte]]) with Serializable {
   private lazy val kryoCodecConstructor: MethodHandle = {
     val cls = SparkClassUtils.classForName(
       "org.apache.spark.sql.catalyst.encoders.KryoSerializationCodecImpl")

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoderSuite.scala

@@ -612,6 +612,7 @@ class ExpressionEncoderSuite extends CodegenInterpretedPlanTest with AnalysisTes
       provider,
       nullable = true))
       .resolveAndBind()
+    assert(encoder.isInstanceOf[Serializable])
     assert(encoder.schema == new StructType().add("value", BinaryType))
     val toRow = encoder.createSerializer()
     val fromRow = encoder.createDeserializer()

sql/core/src/test/scala/org/apache/spark/sql/UDFSuite.scala

@@ -53,6 +53,17 @@ private case class FunctionResult(f1: String, f2: String)
 private case class LocalDateInstantType(date: LocalDate, instant: Instant)
 private case class TimestampInstantType(t: Timestamp, instant: Instant)
 
+private case class KryoEncodedBuf(value: Long)
+private case class KryoBufAggregator() extends Aggregator[Long, KryoEncodedBuf, Long] {
+  override def zero: KryoEncodedBuf = KryoEncodedBuf(0)
+  override def reduce(b: KryoEncodedBuf, a: Long): KryoEncodedBuf = KryoEncodedBuf(b.value + a)
+  override def merge(b1: KryoEncodedBuf, b2: KryoEncodedBuf): KryoEncodedBuf =
+    KryoEncodedBuf(b1.value + b2.value)
+  override def finish(reduction: KryoEncodedBuf): Long = reduction.value
+  override def bufferEncoder: Encoder[KryoEncodedBuf] = Encoders.kryo[KryoEncodedBuf]
+  override def outputEncoder: Encoder[Long] = Encoders.scalaLong
+}
+
 class UDFSuite extends QueryTest with SharedSparkSession {
   import testImplicits._
 
@@ -1249,4 +1260,11 @@ class UDFSuite extends QueryTest with SharedSparkSession {
       .select(f($"c").as("f"), f($"f"))
     checkAnswer(df, Seq(Row(2, 3), Row(null, null)))
   }
+
+  test("SPARK-52819: Support using Kryo to encode BUF in Aggregator") {
+    val kryoBufUDAF = udaf(KryoBufAggregator())
+    val input = Seq(1L, 2L, 3L).toDF("value")
+    val result = input.select(kryoBufUDAF($"value").as("sum"))
+    checkAnswer(result, Row(6L) :: Nil)
+  }
 }