Deserialization issue with Kryo 5 in parallel testing

[nvollmar]

Since updating to Kryo 5 we are experience a weird deserialization failure when running tests parallel on multiple Akka actor systems in the same JVM. We could not reproduce this while running tests sequential nor did this ever happen with Kryo 4.

Unable to find class: uth-store-syncer
Serialization trace:
value (scala.Some)
techInitiator (jaso.commons.security.SecurityCtx)
securityCtx (jaso.general.user.api.UserV0$DrctQryUserWithIdentity)
 com.esotericsoftware.kryo.KryoException: Unable to find class: uth-store-syncer
Serialization trace:
value (scala.Some)
techInitiator (jaso.commons.security.SecurityCtx)
securityCtx (jaso.general.user.api.UserV0$DrctQryUserWithIdentity)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:190)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:159)
	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:691)
	at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:118)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:124)
	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:729)
	at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:125)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:124)
	at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:780)
	at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:134)
Caused by: java.lang.ClassNotFoundException: uth-store-syncer
	at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:604)
	at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
	at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
	at java.base/java.lang.Class.forName0(Native Method)
	at java.base/java.lang.Class.forName(Class.java:416)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:184)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:159)
	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:691)
	at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:118)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:124)

The issue occurs deserializing a plain String, the field serializer tries to load a class by name with the content instead of just using that value. Any insight into what's going wrong would be very helpful.

[nvollmar]

I was able to write a test reproducing those issues here: https://github.com/altoo-ag/akka-kryo-serialization/blob/wip-parallel-serialization-issues-nvo/src/test/scala/io/altoo/akka/serialization/kryo/ParallelActorSystemSerializationTest.scala

[theigl]

@nvollmar: Is any Kryo object or state shared between multiple threads? Since Kryo is not threadsafe, there is a lot that can go wrong in that case.

The fact that the problem goes away when you switch to sequential tests indicates accidental multithreaded access.

[nvollmar]

This was also our first thought, even though we use a pool to guard against multi-threaded access.
I've added some checks and debug statements to verify that and it does not look like the same Kryo instance is accessed concurrently.

We did not change anything in that regard when updating from Kryo 4 to 5, and the same code never had any issues with Kryo 4 in that regard.

[nvollmar]

Interestingly the Option seems to be the important point here. If the field is a simple String I cannot produce any failure, also if I implement and register a custom serializer for the Scala Option class it seems to work just fine.

Now I'm really confused...

[theigl]

I've added some checks and debug statements to verify that and it does not look like the same Kryo instance is accessed concurrently.

Could it be that any other class is shared between threads? One of the resolvers or Input/Output?

We did not change anything in that regard when updating from Kryo 4 to 5, and the same code never had any issues with Kryo 4 in that regard.

The test code might have been incorrect before and the issue simply never manifested itself with Kryo 4.

Interestingly the Option seems to be the important point here. If the field is a simple String I cannot produce any failure, also if I implement and register a custom serializer for the Scala Option class it seems to work just fine.
Now I'm really confused..

Ok that's really strange. Could you try to replicate the problem in a minimal test-case without Akka? It could create a couple of threads and serialize/deserialize the same object your original test uses. If that test passes, there has to be some issue in your current test setup.

[nvollmar]

Ok that's really strange. Could you try to replicate the problem in a minimal test-case without Akka? It could create a couple of threads and serialize/deserialize the same object your original test uses. If that test passes, there has to be some issue in your current test setup.

I managed to write a test-case that reproduces this issue with plain Scala/Kryo:

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{Input, Output}
import com.esotericsoftware.kryo.util._
import org.objenesis.strategy.StdInstantiatorStrategy
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers

import scala.concurrent._
import scala.concurrent.duration._


final case class Sample1(value: Option[String]) extends MySerializable {
  override def toString: String = "sample"
}

class ParallelSerializerTest extends AnyFlatSpec with Matchers {

  private val serializerPool = new Pool[Kryo](true, false, 8) {
    protected def create(): Kryo = {
      val kryo: Kryo = new Kryo(new DefaultClassResolver(), new MapReferenceResolver())
      kryo.setRegistrationRequired(false)
      kryo.setReferences(true)
      val instStrategy = kryo.getInstantiatorStrategy.asInstanceOf[DefaultInstantiatorStrategy]
      instStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy())
      kryo.setInstantiatorStrategy(instStrategy)
      kryo
    }
  }

  private def serializeDeserialize(sample: Sample1): Unit = {
    val sKryo = serializerPool.obtain()
    val obuf = new Output(4096, -1)
    sKryo.writeClassAndObject(obuf, sample)
    val serialized = obuf.toBytes
    obuf.close()
    serializerPool.free(sKryo)

    val dKryo = serializerPool.obtain()
    val ibuf = new Input(serialized)
    val result = dKryo.readClassAndObject(ibuf)
    ibuf.close()
    serializerPool.free(dKryo)
    assert(result == sample)
  }

  it should "serialize and deserialize" in {
    implicit val dispatcher: ExecutionContextExecutor = ExecutionContext.global

    val results1 = for (i <- 0 to 10)
      yield Future {
        serializeDeserialize(Sample1(Some(s"test-$i")))
        serializeDeserialize(Sample1(Some(s"test-$i")))
        serializeDeserialize(Sample1(Some(s"test-$i")))
        serializeDeserialize(Sample1(Some(s"test-$i")))
      }

    Await.result(Future.sequence(results1), 10.seconds)
  }
}

This reproduces the same serialization issue we observed:

Unable to find class: est-2
Serialization trace:
value (scala.Some)
value (io.altoo.akka.serialization.kryo.Sample1)
com.esotericsoftware.kryo.KryoException: Unable to find class: est-2
Serialization trace:
value (scala.Some)
value (io.altoo.akka.serialization.kryo.Sample1)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:190)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:159)
	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:691)
	at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:118)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:124)
	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:729)
	at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:125)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:124)
	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:810)
	at io.altoo.akka.serialization.kryo.ParallelSerializerTest.serializeDeserialize(ParallelSerializerTest.scala:43)
	at io.altoo.akka.serialization.kryo.ParallelSerializerTest.$anonfun$new$3(ParallelSerializerTest.scala:54)
	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
	at scala.concurrent.Future$.$anonfun$apply$1(Future.scala:671)
	at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:430)
	at scala.concurrent.BatchingExecutor$AbstractBatch.runN(BatchingExecutor.scala:134)
	at scala.concurrent.BatchingExecutor$AsyncBatch.apply(BatchingExecutor.scala:163)
	at scala.concurrent.BatchingExecutor$AsyncBatch.apply(BatchingExecutor.scala:146)
	at scala.concurrent.BlockContext$.usingBlockContext(BlockContext.scala:107)
	at scala.concurrent.BatchingExecutor$AsyncBatch.run(BatchingExecutor.scala:154)
	at java.base/java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1425)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1016)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1665)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1598)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:177)
Caused by: java.lang.ClassNotFoundException: est-2
	at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:602)
	at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
	at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
	at java.base/java.lang.Class.forName0(Native Method)
	at java.base/java.lang.Class.forName(Class.java:416)
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:184)
	... 24 more

[theigl]

Thank you very much @nvollmar! I'll try to replicate the test in Java so we can get to the bottom of this.

[theigl]

I rewrote your test in Java and it passes without problems:

package com.esotericsoftware.kryo.serializers;

import static org.junit.Assert.*;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.util.DefaultClassResolver;
import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
import com.esotericsoftware.kryo.util.MapReferenceResolver;
import com.esotericsoftware.kryo.util.Pool;

import java.io.Serializable;
import java.util.Objects;
import java.util.stream.IntStream;

import org.junit.Test;
import org.objenesis.strategy.StdInstantiatorStrategy;

public class ParallelSerializerTest {

	private final Pool<Kryo> serializerPool = new Pool<Kryo>(true, false, 8) {
		@Override
		protected Kryo create () {
			Kryo kryo = new Kryo(new DefaultClassResolver(), new MapReferenceResolver());
			kryo.setRegistrationRequired(false);
			kryo.setReferences(true);
			DefaultInstantiatorStrategy instStrategy = (DefaultInstantiatorStrategy)kryo.getInstantiatorStrategy();
			instStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
			kryo.setInstantiatorStrategy(instStrategy);
			return kryo;
		}
	};

	@Test
	public void serializeAndDeserialize () {
		IntStream.range(0, 1_000)
			.parallel()
			.forEach(i -> serializeDeserialize(new Sample1(new Option<>("test-" + i)), i));
	}

	private void serializeDeserialize (Sample1 sample, int i) {
		System.out.println("Iteration: " + i);

		Kryo sKryo = serializerPool.obtain();
		Output out = new Output(4096, -1);
		sKryo.writeClassAndObject(out, sample);
		byte[] serialized = out.toBytes();
		out.close();
		serializerPool.free(sKryo);

		Kryo dKryo = serializerPool.obtain();
		Input in = new Input(serialized);
		Sample1 result = (Sample1)dKryo.readClassAndObject(in);
		in.close();
		serializerPool.free(dKryo);
		assertEquals(result, sample);
	}

	static final class Option<T> {
		private final T value;

		public Option (T value) {
			this.value = value;
		}

		public T getValue () {
			return value;
		}

		@Override
		public boolean equals (Object o) {
			if (this == o) return true;
			if (o == null || getClass() != o.getClass()) return false;
			Option<?> option = (Option<?>)o;
			return Objects.equals(value, option.value);
		}

		@Override
		public int hashCode () {
			return Objects.hash(value);
		}
	}

	static final class Sample1 implements Serializable {
		private final Option<String> value;

		Sample1 (Option<String> value) {
			this.value = value;
		}

		public Option<String> getValue () {
			return value;
		}

		@Override
		public boolean equals (Object o) {
			if (this == o) return true;
			if (o == null || getClass() != o.getClass()) return false;
			Sample1 sample1 = (Sample1)o;
			return Objects.equals(value, sample1.value);
		}

		@Override
		public int hashCode () {
			return Objects.hash(value);
		}
	}
}

There seems to be something special about the Scala Option class that causes problems with ReflectField.

[nvollmar]

Yes, the issue seems linked to the Scala Option (which would not be a big deal since a custom serializer could deal with that - if I understood the underlying issue).
What is really worrying is the fact that it only happens in highly concurrent scenarios, doing a single serialization run works just fine.

To be noted, the Scala Option is a sealed abstract class, which has two concrete implementations: Some and None.

Did anything fundamentally change between Kryo 4 and 5 in in the area of ReflectField? Since as mentioned, we only experienced this issue with Kryo 5.

[nvollmar]

I can also reproduce the issue with a simplified Option implementation:

sealed abstract class SampleOption[+T]
final case class SampleSome[+T](value: T) extends SampleOption[T]
case object SampleNone extends SampleOption[Nothing]

Supplemental:
Cutting down I find this still fails serialization:

abstract class SampleOption[T]
case class SampleSome[T](value: T) extends SampleOption[T]

Now if I remove the case modifier from SampleSome I cannot get the test to fail anymore.

[nvollmar]

Maybe this is helpful, the disassembled class files:

abstract class SampleOption[T]

javap -p SampleOption.class 
Compiled from "ParallelSerializerTest.scala"
public abstract class io.altoo.akka.serialization.kryo.SampleOption<T> {
  public io.altoo.akka.serialization.kryo.SampleOption();
}

case class SampleSome[T]

> javap -p SampleSome.class

Compiled from "ParallelSerializerTest.scala"
public class io.altoo.akka.serialization.kryo.SampleSome<T> extends io.altoo.akka.serialization.kryo.SampleOption<T> implements scala.Product, java.io.Serializable {
  private final T value;
  public static <T> scala.Option<T> unapply(io.altoo.akka.serialization.kryo.SampleSome<T>);
  public static <T> io.altoo.akka.serialization.kryo.SampleSome<T> apply(T);
  public scala.collection.Iterator<java.lang.String> productElementNames();
  public T value();
  public <T> io.altoo.akka.serialization.kryo.SampleSome<T> copy(T);
  public <T> T copy$default$1();
  public java.lang.String productPrefix();
  public int productArity();
  public java.lang.Object productElement(int);
  public scala.collection.Iterator<java.lang.Object> productIterator();
  public boolean canEqual(java.lang.Object);
  public java.lang.String productElementName(int);
  public int hashCode();
  public java.lang.String toString();
  public boolean equals(java.lang.Object);
  public io.altoo.akka.serialization.kryo.SampleSome(T);
}

Note that a case class comes with a companion object:

javap -p SampleSome$.class
Compiled from "ParallelSerializerTest.scala"
public final class io.altoo.akka.serialization.kryo.SampleSome$ implements java.io.Serializable {
  public static final io.altoo.akka.serialization.kryo.SampleSome$ MODULE$;
  public static {};
  public final java.lang.String toString();
  public <T> io.altoo.akka.serialization.kryo.SampleSome<T> apply(T);
  public <T> scala.Option<T> unapply(io.altoo.akka.serialization.kryo.SampleSome<T>);
  private java.lang.Object writeReplace();
  private io.altoo.akka.serialization.kryo.SampleSome$();
}

class SampleSome[T]

> javap -p SampleSome.class
Compiled from "ParallelSerializerTest.scala"
public class io.altoo.akka.serialization.kryo.SampleSome<T> extends io.altoo.akka.serialization.kryo.SampleOption<T> {
  private final T value;
  public T value();
  public io.altoo.akka.serialization.kryo.SampleSome(T);
}

[nvollmar]

Another finding (sorry for the flood of posts): The issue seems to manifest itself if a case class extends a generic class, removing the type parameter from SampleOption I also cannot the test to fail:

class SampleOption
case class SampleSome[T](value: T) extends SampleOption

[nvollmar]

@theigl I could come up with a pure Java example that fails serialization the same way:

public abstract class JavaOption<T> implements Serializable {

}

class JavaSome<T> extends JavaOption<T> implements java.io.Serializable {
  private final T value;

  public JavaSome(T value) {
    this.value = value;
  }

  public T value() {
    return value;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;
    JavaSome<?> javaSome = (JavaSome<?>) o;
    return Objects.equals(value, javaSome.value);
  }

  @Override
  public int hashCode() {
    return Objects.hash(value);
  }
}

class JavaSample {

  private final JavaOption<String> value;

  public JavaSample(JavaOption<String> value) {
    this.value = value;
  }

  public JavaOption<String> value() {
    return value;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;
    JavaSample that = (JavaSample) o;
    return Objects.equals(value, that.value);
  }

  @Override
  public int hashCode() {
    return Objects.hash(value);
  }
}

Unable to find class: est-2
Serialization trace:
value (io.altoo.akka.serialization.kryo.JavaSome)
value (io.altoo.akka.serialization.kryo.JavaSample)
com.esotericsoftware.kryo.KryoException: Unable to find class: est-2
Serialization trace:
value (io.altoo.akka.serialization.kryo.JavaSome)
value (io.altoo.akka.serialization.kryo.JavaSample)