[FLINK-34996][Connectors/Kafka] Test proper class loader is used and also correct serializer.

Author: hugogu

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

@@ -71,7 +71,7 @@ public void open(InitializationContext context) throws Exception {
                     InstantiationUtil.instantiate(
                             serializerClass.getName(),
                             Serializer.class,
-                            getClass().getClassLoader());
+                            userCodeClassLoader);
 
             if (serializer instanceof Configurable) {
                 ((Configurable) serializer).configure(config);

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

@@ -0,0 +1,28 @@
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.flink.streaming.connectors.kafka.testutils.SerializationTestBase;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaSerializerWrapperTest extends SerializationTestBase {
+    @Override
+    protected void setupContext() {
+        when(serializationContext.getUserCodeClassLoader()).thenReturn(userCodeClassLoader);
+    }
+
+    @Test
+    public void testUserCodeClassLoaderIsUsed() throws Exception {
+        final KafkaSerializerWrapper<String> wrapper =
+                new KafkaSerializerWrapper<>(StringSerializer.class, true, (value) -> "topic");
+
+        testUserClassLoaderIsUsedWhen(() -> {
+            wrapper.open(serializationContext);
+            return null;
+        }, new StringSerializer());
+    }
+}

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

@@ -0,0 +1,32 @@
+package org.apache.flink.connector.kafka.source.reader.deserializer;
+
+import org.apache.flink.streaming.connectors.kafka.testutils.SerializationTestBase;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaValueOnlyDeserializerWrapperTest extends SerializationTestBase {
+    @Override
+    protected void setupContext() {
+        when(deserializationContext.getUserCodeClassLoader()).thenReturn(userCodeClassLoader);
+    }
+
+    @Test
+    public void testUserCodeClassLoaderIsUsed() throws Exception {
+        final Map<String, String> config = new HashMap<>();
+        final KafkaValueOnlyDeserializerWrapper<String> wrapper =
+                new KafkaValueOnlyDeserializerWrapper<>(StringDeserializer.class, config);
+
+        testUserClassLoaderIsUsedWhen(() -> {
+            wrapper.open(deserializationContext);
+            return null;
+        }, new StringDeserializer());
+    }
+}

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/SerializationTestBase.java

@@ -0,0 +1,58 @@
+package org.apache.flink.streaming.connectors.kafka.testutils;
+
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.UserCodeClassLoader;
+import org.junit.Before;
+import org.mockito.*;
+
+import java.util.concurrent.Callable;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.when;
+
+public class SerializationTestBase {
+    @Mock
+    protected DeserializationSchema.InitializationContext deserializationContext;
+    @Mock
+    protected SerializationSchema.InitializationContext serializationContext;
+    @Mock
+    protected UserCodeClassLoader userCodeClassLoader;
+    @Mock
+    protected ClassLoader classLoader;
+    @Captor
+    private ArgumentCaptor<ClassLoader> classLoaderCaptor;
+
+    @Before
+    public void setUp() {
+        when(userCodeClassLoader.asClassLoader()).thenReturn(classLoader);
+        setupContext();
+    }
+
+    protected void setupContext() {
+    }
+
+    protected void testUserClassLoaderIsUsedWhen(Callable<Object> callable, Object instance) throws Exception {
+        try (MockedStatic<InstantiationUtil> mocked = Mockito.mockStatic(InstantiationUtil.class)) {
+
+            mocked.when(() -> InstantiationUtil.instantiate(
+                    anyString(),
+                    notNull(),
+                    any(ClassLoader.class)
+            )).thenReturn(instance);
+
+            callable.call();
+
+            mocked.verify(() -> InstantiationUtil.instantiate(
+                    anyString(),
+                    notNull(),
+                    classLoaderCaptor.capture()
+            ));
+
+            assertEquals(classLoader, classLoaderCaptor.getValue());
+        }
+    }
+}

flink-connector-kafka/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker

@@ -0,0 +1 @@
+mock-maker-inline