[FLINK-34996][Connectors/Kafka] Simplify code and add more tests on serialization wrappers.

Author: hugogu

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java

@@ -63,7 +63,7 @@ class KafkaSerializerWrapper<IN> implements SerializationSchema<IN> {
 
     @Override
     public void open(InitializationContext context) throws Exception {
-        final ClassLoader userCodeClassLoader = selectClassLoader(context);
+        final ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
         try (TemporaryClassLoaderContext ignored =
                 TemporaryClassLoaderContext.of(userCodeClassLoader)) {
             initializeSerializer(userCodeClassLoader);
@@ -84,14 +84,6 @@ public byte[] serialize(IN element) {
         return serializer.serialize(topicSelector.apply(element), element);
     }
 
-    /**
-     * Selects the class loader to be used when instantiating the serializer. Using a class loader
-     * with user code allows users to customize the serializer.
-     */
-    protected ClassLoader selectClassLoader(InitializationContext context) {
-        return context.getUserCodeClassLoader().asClassLoader();
-    }
-
     @SuppressWarnings("unchecked")
     protected void initializeSerializer(ClassLoader classLoader) throws Exception {
         serializer =

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java

@@ -56,7 +56,7 @@ class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserialization
 
     @Override
     public void open(DeserializationSchema.InitializationContext context) throws Exception {
-        ClassLoader userCodeClassLoader = selectClassLoader(context);
+        ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
         try (TemporaryClassLoaderContext ignored =
                 TemporaryClassLoaderContext.of(userCodeClassLoader)) {
             initializeDeserializer(userCodeClassLoader);
@@ -98,14 +98,6 @@ public TypeInformation<T> getProducedType() {
         return TypeExtractor.createTypeInfo(Deserializer.class, deserializerClass, 0, null, null);
     }
 
-    /**
-     * Selects the class loader to be used when instantiating the deserializer. Using a class loader
-     * with user code allows users to customize the deserializer.
-     */
-    protected ClassLoader selectClassLoader(DeserializationSchema.InitializationContext context) {
-        return context.getUserCodeClassLoader().asClassLoader();
-    }
-
     @SuppressWarnings("unchecked")
     protected void initializeDeserializer(ClassLoader classLoader) throws Exception {
         deserializer =

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapperTest.java

@@ -1,6 +1,7 @@
 package org.apache.flink.connector.kafka.sink;
 
 import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.connector.testutils.formats.DummyInitializationContext;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.util.FlinkUserCodeClassLoaders;
@@ -43,6 +44,14 @@ public UserCodeClassLoader getUserCodeClassLoader() {
         assertEquals(classLoader, wrapper.getClassLoaderUsed());
     }
 
+    @Test
+    public void testDefaultClassLoaderIsUsed() throws Exception {
+        final KafkaSerializerWrapperCaptureForTest wrapper = new KafkaSerializerWrapperCaptureForTest();
+        wrapper.open(new DummyInitializationContext());
+
+        assertEquals(DummyInitializationContext.class.getClassLoader(), wrapper.getClassLoaderUsed());
+    }
+
     static class KafkaSerializerWrapperCaptureForTest extends KafkaSerializerWrapper<String> {
         private ClassLoader classLoaderUsed;
 

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapperTest.java

@@ -1,6 +1,7 @@
 package org.apache.flink.connector.kafka.source.reader.deserializer;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.connector.testutils.formats.DummyInitializationContext;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.util.FlinkUserCodeClassLoaders;
@@ -44,6 +45,14 @@ public UserCodeClassLoader getUserCodeClassLoader() {
         assertEquals(classLoader, wrapper.getClassLoaderUsed());
     }
 
+    @Test
+    public void testDefaultClassLoaderIsUsed() throws Exception {
+        final KafkaValueOnlyDeserializerWrapperCaptureForTest wrapper = new KafkaValueOnlyDeserializerWrapperCaptureForTest();
+        wrapper.open(new DummyInitializationContext());
+
+        assertEquals(DummyInitializationContext.class.getClassLoader(), wrapper.getClassLoaderUsed());
+    }
+
     static class KafkaValueOnlyDeserializerWrapperCaptureForTest
             extends KafkaValueOnlyDeserializerWrapper<String> {
         private ClassLoader classLoaderUsed;