fix: Don't use Java serialization for cluster metrics (#32748)

* it was possibly used for BigDecimal and BigInteger
* might result in serialization errors during rolling update, but that should
  heal as soon as the rollout is completed

Author: patriknw

akka-cluster-metrics/src/main/scala/akka/cluster/metrics/Metric.scala

@@ -260,13 +260,15 @@ private[metrics] trait MetricNumericConverter {
    * May involve rounding or truncation.
    */
   def convertNumber(from: Any): Either[Long, Double] = from match {
-    case n: Int        => Left(n)
-    case n: Long       => Left(n)
-    case n: Double     => Right(n)
-    case n: Float      => Right(n)
-    case n: BigInt     => Left(n.longValue)
-    case n: BigDecimal => Right(n.doubleValue)
-    case x             => throw new IllegalArgumentException(s"Not a number [$x]")
+    case n: Int                  => Left(n)
+    case n: Long                 => Left(n)
+    case n: Double               => Right(n)
+    case n: Float                => Right(n)
+    case n: BigInt               => Left(n.longValue)
+    case n: java.math.BigInteger => Left(n.longValue)
+    case n: BigDecimal           => Right(n.doubleValue)
+    case n: java.math.BigDecimal => Right(n.doubleValue)
+    case x                       => throw new IllegalArgumentException(s"Not a number [$x]")
   }
 
 }

akka-cluster-metrics/src/main/scala/akka/cluster/metrics/protobuf/MessageSerializer.scala

@@ -22,6 +22,8 @@ import akka.serialization.{ BaseSerializer, SerializationExtension, SerializerWi
 import akka.util.ClassLoaderObjectInputStream
 import scala.jdk.CollectionConverters._
 
+import akka.serialization.DisabledJavaSerializer
+
 /**
  * Protobuf serializer for [[akka.cluster.metrics.ClusterMetricsMessage]] types.
  */
@@ -186,23 +188,35 @@ class MessageSerializer(val system: ExtendedActorSystem) extends SerializerWithS
       cm.NodeMetrics.EWMA.newBuilder().setValue(x.value).setAlpha(x.alpha)
     }
 
-    def numberToProto(number: Number): cm.NodeMetrics.Number.Builder = {
+    @tailrec def numberToProto(number: Number): cm.NodeMetrics.Number.Builder = {
       import cm.NodeMetrics.Number
       import cm.NodeMetrics.NumberType
       number match {
         case n: jl.Double  => Number.newBuilder().setType(NumberType.Double).setValue64(jl.Double.doubleToLongBits(n))
         case n: jl.Long    => Number.newBuilder().setType(NumberType.Long).setValue64(n)
         case n: jl.Float   => Number.newBuilder().setType(NumberType.Float).setValue32(jl.Float.floatToIntBits(n))
         case n: jl.Integer => Number.newBuilder().setType(NumberType.Integer).setValue32(n)
+        case n: java.math.BigInteger =>
+          numberToProto(n.longValue) // this truncation is anyway in MetricNumericConverter
+        case n: BigInt => numberToProto(n.toLong) // this truncation is anyway in MetricNumericConverter
+        case n: java.math.BigDecimal =>
+          numberToProto(n.doubleValue) // this rounding is anyway in MetricNumericConverter
+        case n: BigDecimal => numberToProto(n.toDouble) // this rounding is anyway in MetricNumericConverter
         case _ =>
-          val bos = new ByteArrayOutputStream
-          val out = new ObjectOutputStream(bos)
-          out.writeObject(number)
-          out.close()
-          Number
-            .newBuilder()
-            .setType(NumberType.Serialized)
-            .setSerialized(ByteStringUtils.toProtoByteStringUnsafe(bos.toByteArray))
+          if (system.settings.AllowJavaSerialization) {
+            val bos = new ByteArrayOutputStream
+            val out = new ObjectOutputStream(bos)
+            out.writeObject(number)
+            out.close()
+            Number
+              .newBuilder()
+              .setType(NumberType.Serialized)
+              .setSerialized(ByteStringUtils.toProtoByteStringUnsafe(bos.toByteArray))
+          } else {
+            // this is the default, and it shouldn't happen since all number types should be covered above
+            throw throw new DisabledJavaSerializer.JavaSerializationException(
+              s"Unsupported number [${number.getClass.getName}], when Java serialization is disabled")
+          }
       }
     }
 
@@ -253,12 +267,18 @@ class MessageSerializer(val system: ExtendedActorSystem) extends SerializerWithS
         case NumberType.Float_VALUE   => jl.Float.intBitsToFloat(number.getValue32)
         case NumberType.Integer_VALUE => number.getValue32
         case NumberType.Serialized_VALUE =>
-          val in = new ClassLoaderObjectInputStream(
-            system.dynamicAccess.classLoader,
-            new ByteArrayInputStream(number.getSerialized.toByteArray))
-          val obj = in.readObject
-          in.close()
-          obj.asInstanceOf[jl.Number]
+          if (system.settings.AllowJavaSerialization) {
+            val in = new ClassLoaderObjectInputStream(
+              system.dynamicAccess.classLoader,
+              new ByteArrayInputStream(number.getSerialized.toByteArray))
+            val obj = in.readObject
+            in.close()
+            obj.asInstanceOf[jl.Number]
+          } else {
+            // this is the default, and it shouldn't happen since all number types should be covered above
+            throw throw new DisabledJavaSerializer.JavaSerializationException(
+              s"Unsupported number [${number.getClass.getName}], when Java serialization is disabled")
+          }
       }
     }
 

akka-cluster-metrics/src/test/scala/akka/cluster/metrics/protobuf/MessageSerializerSpec.scala

@@ -4,6 +4,8 @@
 
 package akka.cluster.metrics.protobuf
 
+import java.math.BigInteger
+
 import akka.actor.{ Address, ExtendedActorSystem }
 import akka.cluster.MemberStatus
 import akka.cluster.TestMember
@@ -52,7 +54,10 @@ class MessageSerializerSpec extends AkkaSpec("""
               Metric("bar2", Float.MaxValue, None),
               Metric("bar3", Int.MaxValue, None),
               Metric("bar4", Long.MaxValue, None),
-              Metric("bar5", BigInt(Long.MaxValue), None)))))
+              Metric("bar5", BigInt(Long.MaxValue), None),
+              Metric("bar6", new BigInteger(s"${Long.MaxValue}"), None),
+              Metric("bar7", BigDecimal.exact(s"${Long.MaxValue}.0"), None),
+              Metric("bar8", new java.math.BigDecimal(s"${Long.MaxValue}.0"), None)))))
 
       checkSerialization(MetricsGossipEnvelope(a1.address, metricsGossip, true))