KAFKA-16921: Migrate test of connect module to Junit5 (Runtime direct)

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java

@@ -56,11 +56,14 @@
 import org.apache.kafka.connect.util.Callback;
 import org.apache.kafka.connect.util.ConnectorTaskId;
 import org.apache.kafka.connect.util.FutureCallback;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -79,12 +82,12 @@
 import java.util.stream.Collectors;
 
 import static org.apache.kafka.connect.runtime.AbstractHerder.keysWithVariableValues;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -98,7 +101,8 @@
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.withSettings;
 
-@RunWith(MockitoJUnitRunner.StrictStubs.class)
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
 public class AbstractHerderTest {
 
     private static final String CONN1 = "sourceA";
@@ -455,6 +459,7 @@ public void testConfigValidationEmptyConfig() {
 
         assertThrows(BadRequestException.class, () -> herder.validateConnectorConfig(Collections.emptyMap(), s -> null, false));
         verify(transformer).transform(Collections.emptyMap());
+        assertEquals(worker.getPlugins(), plugins);
     }
 
     @Test
@@ -1050,24 +1055,24 @@ private <T> void testConnectorPluginConfig(
         verify(plugins).withClassLoader(newPluginInstance.get().getClass().getClassLoader());
     }
 
-    @Test(expected = NotFoundException.class)
+    @Test
     public void testGetConnectorConfigDefWithBadName() throws Exception {
         String connName = "AnotherPlugin";
         AbstractHerder herder = testHerder();
         when(worker.getPlugins()).thenReturn(plugins);
         when(plugins.pluginClass(anyString())).thenThrow(new ClassNotFoundException());
-        herder.connectorPluginConfig(connName);
+        assertThrows(NotFoundException.class, () -> herder.connectorPluginConfig(connName));
     }
 
-    @Test(expected = BadRequestException.class)
+    @Test
     @SuppressWarnings({"rawtypes", "unchecked"})
     public void testGetConnectorConfigDefWithInvalidPluginType() throws Exception {
         String connName = "AnotherPlugin";
         AbstractHerder herder = testHerder();
         when(worker.getPlugins()).thenReturn(plugins);
         when(plugins.pluginClass(anyString())).thenReturn((Class) Object.class);
         when(plugins.newPlugin(anyString())).thenReturn(new DirectoryConfigProvider());
-        herder.connectorPluginConfig(connName);
+        assertThrows(BadRequestException.class, () -> herder.connectorPluginConfig(connName));
     }
 
     @Test
@@ -1228,7 +1233,7 @@ private void testConfigProviderRegex(String rawConnConfig) {
     private void testConfigProviderRegex(String rawConnConfig, boolean expected) {
         Set<String> keys = keysWithVariableValues(Collections.singletonMap("key", rawConnConfig), ConfigTransformer.DEFAULT_PATTERN);
         boolean actual = keys != null && !keys.isEmpty() && keys.contains("key");
-        assertEquals(String.format("%s should have matched regex", rawConnConfig), expected, actual);
+        assertEquals(expected, actual, String.format("%s should have matched regex", rawConnConfig));
     }
 
     private AbstractHerder createConfigValidationHerder(Class<? extends Connector> connectorClass,

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java

@@ -55,14 +55,17 @@
 import org.apache.kafka.connect.util.ConnectorTaskId;
 import org.apache.kafka.connect.util.TopicAdmin;
 import org.apache.kafka.connect.util.TopicCreationGroup;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.AdditionalAnswers;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
 import org.mockito.stubbing.Answer;
 
 import java.nio.ByteBuffer;
@@ -90,15 +93,16 @@
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.PARTITIONS_CONFIG;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_FACTOR_CONFIG;
 import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_CREATION_ENABLE_CONFIG;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -107,7 +111,8 @@
 import static org.mockito.Mockito.when;
 
 @SuppressWarnings("unchecked")
-@RunWith(MockitoJUnitRunner.StrictStubs.class)
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
 public class AbstractWorkerSourceTaskTest {
 
     private static final String TOPIC = "topic";
@@ -151,7 +156,7 @@ public class AbstractWorkerSourceTaskTest {
 
     private AbstractWorkerSourceTask workerTask;
 
-    @Before
+    @BeforeEach
     public void setup() {
         Map<String, String> workerProps = workerProps();
         plugins = new Plugins(workerProps);
@@ -187,7 +192,7 @@ private Map<String, String> sourceConnectorPropsWithGroups() {
         return props;
     }
 
-    @After
+    @AfterEach
     public void tearDown() {
         if (metrics != null) metrics.stop();
         verifyNoMoreInteractions(statusListener);
@@ -249,6 +254,7 @@ public void testSendRecordsConvertsData() {
         );
 
         expectSendRecord(emptyHeaders());
+        expectApplyTransformationChain();
         expectTopicCreation(TOPIC);
 
         workerTask.toSend = records;
@@ -269,6 +275,7 @@ public void testSendRecordsPropagatesTimestamp() {
         createWorkerTask();
 
         expectSendRecord(emptyHeaders());
+        expectApplyTransformationChain();
         expectTopicCreation(TOPIC);
 
         workerTask.toSend = Collections.singletonList(
@@ -288,7 +295,8 @@ public void testSendRecordsCorruptTimestamp() {
         final Long timestamp = -3L;
         createWorkerTask();
 
-        expectSendRecord(emptyHeaders());
+        expectConvertHeadersAndKeyValue(emptyHeaders(), TOPIC);
+        expectApplyTransformationChain();
 
         workerTask.toSend = Collections.singletonList(
                 new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp)
@@ -304,6 +312,7 @@ public void testSendRecordsNoTimestamp() {
         createWorkerTask();
 
         expectSendRecord(emptyHeaders());
+        expectApplyTransformationChain();
         expectTopicCreation(TOPIC);
 
         workerTask.toSend = Collections.singletonList(
@@ -329,6 +338,7 @@ public void testHeaders() {
         createWorkerTask();
 
         expectSendRecord(headers);
+        expectApplyTransformationChain();
         expectTopicCreation(TOPIC);
 
         workerTask.toSend = Collections.singletonList(
@@ -356,6 +366,7 @@ public void testHeadersWithCustomConverter() throws Exception {
                 Collections::emptyList);
 
         expectSendRecord(null);
+        expectApplyTransformationChain();
         expectTopicCreation(TOPIC);
 
         String stringA = "Árvíztűrő tükörfúrógép";
@@ -617,6 +628,7 @@ public void testTopicCreateSucceedsWhenCreateReturnsExistingTopicFound() {
         SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
 
         expectSendRecord(emptyHeaders());
+        expectApplyTransformationChain();
 
         when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap());
         when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(foundTopic(TOPIC));
@@ -641,6 +653,7 @@ public void testTopicCreateSucceedsWhenCreateReturnsNewTopicFound() {
         SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
 
         expectSendRecord(emptyHeaders());
+        expectApplyTransformationChain();
 
         when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap());
         when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC));
@@ -666,7 +679,6 @@ public void testSendRecordsRetriableException() {
         SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
 
         expectConvertHeadersAndKeyValue(emptyHeaders(), TOPIC);
-        expectTaskGetTopic();
 
         when(transformationChain.apply(any(), eq(record1))).thenReturn(null);
         when(transformationChain.apply(any(), eq(record2))).thenReturn(null);
@@ -698,8 +710,6 @@ private void expectSendRecord(Headers headers) {
         if (headers != null)
             expectConvertHeadersAndKeyValue(headers, TOPIC);
 
-        expectApplyTransformationChain();
-
         expectTaskGetTopic();
     }
 
@@ -792,11 +802,15 @@ private void expectConvertHeadersAndKeyValue(Headers headers, String topic) {
         when(valueConverter.fromConnectData(eq(topic), any(Headers.class), eq(RECORD_SCHEMA),
                 eq(RECORD)))
                 .thenReturn(SERIALIZED_RECORD);
+        assertEquals(keyConverter.fromConnectData(topic, headers, KEY_SCHEMA, KEY), SERIALIZED_KEY);
+        assertEquals(valueConverter.fromConnectData(topic, headers, RECORD_SCHEMA, RECORD), SERIALIZED_RECORD);
     }
 
     private void expectApplyTransformationChain() {
         when(transformationChain.apply(any(), any(SourceRecord.class)))
                 .thenAnswer(AdditionalAnswers.returnsSecondArg());
+        SourceRecord randomString = mock(SourceRecord.class);
+        assertEquals(transformationChain.apply(null, randomString), randomString);
     }
 
     private RecordHeaders emptyHeaders() {

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectMetricsTest.java

@@ -24,21 +24,22 @@
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup;
 import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroupId;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNotSame;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class ConnectMetricsTest {
 
@@ -51,12 +52,12 @@ public class ConnectMetricsTest {
 
     private ConnectMetrics metrics;
 
-    @Before
+    @BeforeEach
     public void setUp() {
         metrics = new ConnectMetrics("worker1", new WorkerConfig(WorkerConfig.baseConfigDef(), DEFAULT_WORKER_CONFIG), new MockTime(), "cluster-1");
     }
 
-    @After
+    @AfterEach
     public void tearDown() {
         if (metrics != null)
             metrics.stop();

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectorConfigTest.java

@@ -26,20 +26,21 @@
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.transforms.Transformation;
 import org.apache.kafka.connect.transforms.predicates.Predicate;
-import org.junit.Test;
+
+import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
 
 public class ConnectorConfigTest<R extends ConnectRecord<R>> {
 
@@ -469,8 +470,8 @@ public void testEnrichedConfigDef() {
     private static void assertEnrichedConfigDef(ConfigDef def, String prefix, String keyName, ConfigDef.Type expectedType) {
         assertNull(def.configKeys().get(keyName));
         ConfigDef.ConfigKey configKey = def.configKeys().get(prefix + keyName);
-        assertNotNull(prefix + keyName + "' config must be present", configKey);
-        assertEquals(prefix + keyName + "' config should be a " + expectedType, expectedType, configKey.type);
+        assertNotNull(configKey, prefix + keyName + "' config must be present");
+        assertEquals(expectedType, configKey.type, prefix + keyName + "' config should be a " + expectedType);
     }
 
     public static class HasDuplicateConfigTransformation<R extends ConnectRecord<R>> implements Transformation<R>, Versioned {