Serializers { public static void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen) { if (typeInfo instanceof GenericTypeInfo) { GenericTypeInfo<?> genericTypeInfo = (GenericTypeInfo<?>) typeInfo; Serializers.recursivelyRegisterType(genericTypeInfo.getTypeClass(), config, alreadySeen); } else if (typeInfo instanceof CompositeType) { List<GenericTypeInfo<?>> genericTypesInComposite = new ArrayList<>(); getContainedGenericTypes((CompositeType<?>)typeInfo, genericTypesInComposite); for (GenericTypeInfo<?> gt : genericTypesInComposite) { Serializers.recursivelyRegisterType(gt.getTypeClass(), config, alreadySeen); } } else if (typeInfo instanceof ObjectArrayTypeInfo) { ObjectArrayTypeInfo<?, ?> objectArrayTypeInfo = (ObjectArrayTypeInfo<?, ?>) typeInfo; recursivelyRegisterType(objectArrayTypeInfo.getComponentInfo(), config, alreadySeen); } }  static void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen); static void recursivelyRegisterType(Class<?> type, ExecutionConfig config, Set<Class<?>> alreadySeen);  }
@Test public void testTypeRegistration() { ExecutionConfig conf = new ExecutionConfig(); Serializers.recursivelyRegisterType(ClassWithNested.class, conf, new HashSet<Class<?>>()); KryoSerializer<String> kryo = new KryoSerializer<>(String.class, conf); Assert.assertTrue(kryo.getKryo().getRegistration(FromNested.class).getId() > 0); Assert.assertTrue(kryo.getKryo().getRegistration(ClassWithNested.class).getId() > 0); Assert.assertTrue(kryo.getKryo().getRegistration(Path.class).getId() > 0); Assert.assertTrue(kryo.getKryo().getRegistration(FromGeneric1.class).getId() > 0); Assert.assertTrue(kryo.getKryo().getRegistration(FromGeneric2.class).getId() > 0); Assert.assertTrue(kryo.getKryo().getRegistration(Node.class).getId() > 0); ExecutionConfig conf2 = new ExecutionConfig(); Serializers.recursivelyRegisterType(ClassWithNested.class, conf2, new HashSet<Class<?>>()); KryoSerializer<String> kryo2 = new KryoSerializer<>(String.class, conf); assertTrue(kryo2.getKryo().getRegistration(FromNested.class).getId() > 0); }
@Test public void testTypeRegistrationFromTypeInfo() { ExecutionConfig conf = new ExecutionConfig(); Serializers.recursivelyRegisterType(new GenericTypeInfo<>(ClassWithNested.class), conf, new HashSet<Class<?>>()); KryoSerializer<String> kryo = new KryoSerializer<>(String.class, conf); assertTrue(kryo.getKryo().getRegistration(FromNested.class).getId() > 0); assertTrue(kryo.getKryo().getRegistration(ClassWithNested.class).getId() > 0); assertTrue(kryo.getKryo().getRegistration(Path.class).getId() > 0); assertTrue(kryo.getKryo().getRegistration(FromGeneric1.class).getId() > 0); assertTrue(kryo.getKryo().getRegistration(FromGeneric2.class).getId() > 0); assertTrue(kryo.getKryo().getRegistration(Node.class).getId() > 0); }