test_class.file: flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/SerializersTest.java
test_class.identifier: SerializersTest
test_case.body: @Test
  	public void testTypeRegistration() {
  		ExecutionConfig conf = new ExecutionConfig();
  		Serializers.recursivelyRegisterType(ClassWithNested.class, conf, new HashSet<Class<?>>());
  		
  		KryoSerializer<String> kryo = new KryoSerializer<>(String.class, conf); // we create Kryo from another type.
  
  		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);
  		
  		// check if the generic type from one field is also registered (its very likely that
  		// generic types are also used as fields somewhere.
  		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);
  		
  		
  		// register again and make sure classes are still registered
  		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_case.class_method_signature: SerializersTest.testTypeRegistration()
test_case.full_signature: @Test public void testTypeRegistration()
test_case.identifier: testTypeRegistration
test_case.invocations: recursivelyRegisterType assertTrue getId getRegistration getKryo assertTrue getId getRegistration getKryo assertTrue getId getRegistration getKryo assertTrue getId getRegistration getKryo assertTrue getId getRegistration getKryo assertTrue getId getRegistration getKryo recursivelyRegisterType assertTrue getId getRegistration getKryo
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testTypeRegistration()
focal_class.file: flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
focal_class.identifier: Serializers
focal_class.class_method_signature: Serializers.recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.full_signature: public static void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.identifier: recursivelyRegisterType
focal_class.modifiers: public static
focal_class.parameters: (TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.return: void
focal_class.signature: void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.class_method_signature: Serializers.recursivelyRegisterType(Class<?> type, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.full_signature: public static void recursivelyRegisterType(Class<?> type, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.identifier: recursivelyRegisterType
focal_class.modifiers: public static
focal_class.parameters: (Class<?> type, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.return: void
focal_class.signature: void recursivelyRegisterType(Class<?> type, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.class_method_signature: Serializers.recursivelyRegisterGenericType(Type fieldType, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.full_signature: private static void recursivelyRegisterGenericType(Type fieldType, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.identifier: recursivelyRegisterGenericType
focal_class.modifiers: private static
focal_class.parameters: (Type fieldType, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.return: void
focal_class.signature: void recursivelyRegisterGenericType(Type fieldType, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_class.class_method_signature: Serializers.getContainedGenericTypes(CompositeType<?> typeInfo, List<GenericTypeInfo<?>> target)
focal_class.full_signature: private static void getContainedGenericTypes(CompositeType<?> typeInfo, List<GenericTypeInfo<?>> target)
focal_class.identifier: getContainedGenericTypes
focal_class.modifiers: private static
focal_class.parameters: (CompositeType<?> typeInfo, List<GenericTypeInfo<?>> target)
focal_class.return: void
focal_class.signature: void getContainedGenericTypes(CompositeType<?> typeInfo, List<GenericTypeInfo<?>> target)
focal_class.class_method_signature: Serializers.checkAndAddSerializerForTypeAvro(ExecutionConfig reg, Class<?> type)
focal_class.full_signature: private static void checkAndAddSerializerForTypeAvro(ExecutionConfig reg, Class<?> type)
focal_class.identifier: checkAndAddSerializerForTypeAvro
focal_class.modifiers: private static
focal_class.parameters: (ExecutionConfig reg, Class<?> type)
focal_class.return: void
focal_class.signature: void checkAndAddSerializerForTypeAvro(ExecutionConfig reg, Class<?> type)
focal_method.body: 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);
  		}
  	}
focal_method.class_method_signature: Serializers.recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_method.full_signature: public static void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_method.identifier: recursivelyRegisterType
focal_method.invocations: recursivelyRegisterType getTypeClass getContainedGenericTypes recursivelyRegisterType getTypeClass recursivelyRegisterType getComponentInfo
focal_method.modifiers: public static
focal_method.parameters: (TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)
focal_method.return: void
focal_method.signature: void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen)