diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java
index cf07d5d..9ced632 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java
@@ -18,6 +18,7 @@
 
 import org.apache.kafka.connect.connector.Connector;
 import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.health.ConnectorType;
 import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup;
 import org.apache.kafka.connect.runtime.isolation.Plugins;
 import org.apache.kafka.connect.sink.SinkConnector;
@@ -29,15 +30,22 @@
 import org.apache.kafka.connect.util.Callback;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
 
+
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+import org.mockito.quality.Strictness;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -54,7 +62,7 @@
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
-@RunWith(MockitoJUnitRunner.StrictStubs.class)
+@RunWith(Parameterized.class)
 public class WorkerConnectorTest {
 
     private static final String VERSION = "1.1";
@@ -68,15 +76,41 @@
     public ConnectorConfig connectorConfig;
     public MockConnectMetrics metrics;
 
+    @Rule
+    public MockitoRule rule = MockitoJUnit.rule().strictness(Strictness.STRICT_STUBS);
+
     @Mock private Plugins plugins;
-    @Mock private SourceConnector sourceConnector;
-    @Mock private SinkConnector sinkConnector;
     @Mock private CloseableConnectorContext ctx;
     @Mock private ConnectorStatus.Listener listener;
-    @Mock private CloseableOffsetStorageReader offsetStorageReader;
-    @Mock private ConnectorOffsetBackingStore offsetStore;
     @Mock private ClassLoader classLoader;
-    private Connector connector;
+
+    private final ConnectorType connectorType;
+    private final Connector connector;
+    private final CloseableOffsetStorageReader offsetStorageReader;
+    private final ConnectorOffsetBackingStore offsetStore;
+
+    @Parameterized.Parameters
+    public static Collection<ConnectorType> parameters() {
+        return Arrays.asList(ConnectorType.SOURCE, ConnectorType.SINK);
+    }
+
+    public WorkerConnectorTest(ConnectorType connectorType) {
+        this.connectorType = connectorType;
+        switch (connectorType) {
+            case SINK:
+                this.connector = mock(SinkConnector.class);
+                this.offsetStorageReader = null;
+                this.offsetStore = null;
+                break;
+            case SOURCE:
+                this.connector = mock(SourceConnector.class);
+                this.offsetStorageReader = mock(CloseableOffsetStorageReader.class);
+                this.offsetStore = mock(ConnectorOffsetBackingStore.class);
+                break;
+            default:
+                throw new IllegalStateException("Unexpected connector type: " + connectorType);
+        }
+    }
 
     @Before
     public void setup() {
@@ -92,7 +126,6 @@
     @Test
     public void testInitializeFailure() {
         RuntimeException exception = new RuntimeException();
-        connector = sourceConnector;
 
         when(connector.version()).thenReturn(VERSION);
         doThrow(exception).when(connector).initialize(any());
@@ -113,13 +146,12 @@
     @Test
     public void testFailureIsFinalState() {
         RuntimeException exception = new RuntimeException();
-        connector = sinkConnector;
 
         when(connector.version()).thenReturn(VERSION);
         doThrow(exception).when(connector).initialize(any());
 
         Callback<TargetState> onStateChange = mockCallback();
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
         assertFailedMetric(workerConnector);
@@ -140,15 +172,13 @@
 
     @Test
     public void testStartupAndShutdown() {
-        connector = sourceConnector;
-
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
         workerConnector.shutdown();
@@ -166,14 +196,12 @@
 
     @Test
     public void testStartupAndPause() {
-        connector = sinkConnector;
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSinkMetric(workerConnector);
 
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
@@ -197,14 +225,13 @@
 
     @Test
     public void testStartupAndStop() {
-        connector = sinkConnector;
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSinkMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
 
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
@@ -228,8 +255,6 @@
 
     @Test
     public void testOnResume() {
-        connector = sourceConnector;
-
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
@@ -237,7 +262,7 @@
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.PAUSED, onStateChange);
         assertPausedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
@@ -260,14 +285,13 @@
 
     @Test
     public void testStartupPaused() {
-        connector = sinkConnector;
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSinkMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.PAUSED, onStateChange);
         assertPausedMetric(workerConnector);
         workerConnector.shutdown();
@@ -285,14 +309,13 @@
 
     @Test
     public void testStartupStopped() {
-        connector = sinkConnector;
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSinkMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STOPPED, onStateChange);
         assertStoppedMetric(workerConnector);
         workerConnector.shutdown();
@@ -311,16 +334,15 @@
     @Test
     public void testStartupFailure() {
         RuntimeException exception = new RuntimeException();
-        connector = sinkConnector;
 
         when(connector.version()).thenReturn(VERSION);
         doThrow(exception).when(connector).start(CONFIG);
 
         Callback<TargetState> onStateChange = mockCallback();
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSinkMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertFailedMetric(workerConnector);
         workerConnector.shutdown();
@@ -339,7 +361,6 @@
     @Test
     public void testStopFailure() {
         RuntimeException exception = new RuntimeException();
-        connector = sourceConnector;
 
         when(connector.version()).thenReturn(VERSION);
 
@@ -352,7 +373,7 @@
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onFirstStateChange);
         assertRunningMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STOPPED, onSecondStateChange);
@@ -382,7 +403,6 @@
     @Test
     public void testShutdownFailure() {
         RuntimeException exception = new RuntimeException();
-        connector = sourceConnector;
 
         when(connector.version()).thenReturn(VERSION);
 
@@ -392,7 +412,7 @@
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
         workerConnector.shutdown();
@@ -410,8 +430,6 @@
 
     @Test
     public void testTransitionStartedToStarted() {
-        connector = sourceConnector;
-
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
@@ -419,7 +437,7 @@
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
@@ -439,14 +457,13 @@
 
     @Test
     public void testTransitionPausedToPaused() {
-        connector = sourceConnector;
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.PAUSED, onStateChange);
@@ -471,14 +488,13 @@
 
     @Test
     public void testTransitionStoppedToStopped() {
-        connector = sourceConnector;
         when(connector.version()).thenReturn(VERSION);
 
         Callback<TargetState> onStateChange = mockCallback();
         WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
-        assertInitializedSourceMetric(workerConnector);
+        assertInitializedMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STARTED, onStateChange);
         assertRunningMetric(workerConnector);
         workerConnector.doTransitionTo(TargetState.STOPPED, onStateChange);
@@ -503,13 +519,13 @@
 
     @Test
     public void testFailConnectorThatIsNeitherSourceNorSink() {
-        connector = mock(Connector.class);
-        when(connector.version()).thenReturn(VERSION);
-        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
+        Connector badConnector = mock(Connector.class);
+        when(badConnector.version()).thenReturn(VERSION);
+        WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, badConnector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader);
 
         workerConnector.initialize();
 
-        verify(connector).version();
+        verify(badConnector).version();
         ArgumentCaptor<Throwable> exceptionCapture = ArgumentCaptor.forClass(Throwable.class);
         verify(listener).onFailure(eq(CONNECTOR), exceptionCapture.capture());
         Throwable e = exceptionCapture.getValue();
@@ -557,12 +573,19 @@
         assertFalse(workerConnector.metrics().isRunning());
     }
 
-    protected void assertInitializedSinkMetric(WorkerConnector workerConnector) {
-        assertInitializedMetric(workerConnector, "sink");
-    }
-
-    protected void assertInitializedSourceMetric(WorkerConnector workerConnector) {
-        assertInitializedMetric(workerConnector, "source");
+    protected void assertInitializedMetric(WorkerConnector workerConnector) {
+        String expectedType;
+        switch (connectorType) {
+            case SINK:
+                expectedType = "sink";
+                break;
+            case SOURCE:
+                expectedType = "source";
+                break;
+            default:
+                throw new IllegalStateException("Unexpected connector type: " + connectorType);
+        }
+        assertInitializedMetric(workerConnector, expectedType);
     }
 
     protected void assertInitializedMetric(WorkerConnector workerConnector, String expectedType) {
@@ -588,10 +611,10 @@
 
     private void verifyInitialize() {
         verify(connector).version();
-        if (connector instanceof SourceConnector) {
+        if (connectorType == ConnectorType.SOURCE) {
             verify(offsetStore).start();
             verify(connector).initialize(any(SourceConnectorContext.class));
-        } else {
+        } else if (connectorType == ConnectorType.SINK) {
             verify(connector).initialize(any(SinkConnectorContext.class));
         }
     }
@@ -606,7 +629,7 @@
 
     private void verifyShutdown(int connectorStops, boolean clean, boolean started) {
         verify(ctx).close();
-        if (connector instanceof SourceConnector) {
+        if (connectorType == ConnectorType.SOURCE) {
             verify(offsetStorageReader).close();
             verify(offsetStore).stop();
         }