[SPARK-54489][CORE] Avoid returning corrupted Kryo objects

[cxzl25]

What changes were proposed in this pull request?

When Kryo.writeClassAndObject throws a NegativeArraySizeException, the Kryo instance shall no longer be returned to the pool.

Why are the changes needed?

https://github.com/EsotericSoftware/kryo/pull/1198

When Kryo.writeClassAndObject fails to resize, it throws a NegativeArraySizeException.
Spark then returns this Kryo instance to the pool. The next time this object is reused, it will always throw an ArrayIndexOutOfBoundsException.

java.lang.NegativeArraySizeException: -2147483645
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.resize(IdentityObjectIntMap.java:542)
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.putStash(IdentityObjectIntMap.java:306)
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.push(IdentityObjectIntMap.java:300)
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.put(IdentityObjectIntMap.java:162)
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.putStash(IdentityObjectIntMap.java:307)
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.push(IdentityObjectIntMap.java:300)
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.put(IdentityObjectIntMap.java:162)
        at com.esotericsoftware.kryo.util.MapReferenceResolver.addWrittenObject(MapReferenceResolver.java:41)
        at com.esotericsoftware.kryo.Kryo.writeReferenceOrNull(Kryo.java:681)
        at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:646)
        at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.write(DefaultArraySerializers.java:361)
        at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.write(DefaultArraySerializers.java:302)
        at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:651)
        at org.apache.spark.serializer.KryoSerializationStream.writeObject(KryoSerializer.scala:269)
        at org.apache.spark.broadcast.TorrentBroadcast$.$anonfun$blockifyObject$4(TorrentBroadcast.scala:358)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1510)
        at org.apache.spark.broadcast.TorrentBroadcast$.blockifyObject(TorrentBroadcast.scala:360)

java.lang.ArrayIndexOutOfBoundsException: Index 688291177 out of bounds for length 3
        at com.esotericsoftware.kryo.util.IdentityObjectIntMap.get(IdentityObjectIntMap.java:322)
        at com.esotericsoftware.kryo.util.MapReferenceResolver.getWrittenId(MapReferenceResolver.java:46)
        at com.esotericsoftware.kryo.Kryo.writeReferenceOrNull(Kryo.java:671)
        at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:646)
        at org.apache.spark.serializer.KryoSerializationStream.writeObject(KryoSerializer.scala:269)
        at org.apache.spark.broadcast.TorrentBroadcast$.$anonfun$blockifyObject$4(TorrentBroadcast.scala:358)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1510)
        at org.apache.spark.broadcast.TorrentBroadcast$.blockifyObject(TorrentBroadcast.scala:360)

Simple test

    IdentityObjectIntMap identityObjectIntMap = new IdentityObjectIntMap(1073741824, 0.8f);
    try {
      identityObjectIntMap.put("k1", 1);
      identityObjectIntMap.clear((1073741824) << 1); // Simulate resize
    } catch (NegativeArraySizeException e) {  
      e.printStackTrace(); // Expected
    }
    identityObjectIntMap.clear(2048);
    identityObjectIntMap.put("k1", 1); // ArrayIndexOutOfBoundsException

Does this PR introduce any user-facing change?

No

How was this patch tested?

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

No

[holdenk]

Is it possible to add a test here?