Repository: apache/kafka
File: connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java (Java)

Code before:

    private void verifyTaskHeaderConverter() {
        verify(plugins).newHeaderConverter(any(AbstractConfig.class), eq(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(ClassLoaderUsage.CURRENT_CLASSLOADER));
    }

    private void mockVersionedTaskHeaderConverterFromConnector(HeaderConverter returning) {
        when(plugins.newHeaderConverter(any(ConnectorConfig.class), eq(ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(ConnectorConfig.HEADER_CONVERTER_VERSION_CONFIG)))
               .thenReturn(returning);
    }

    private void verifyVersionedTaskHeaderConverterFromConnector() {
        verify(plugins).newHeaderConverter(any(ConnectorConfig.class), eq(ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(ConnectorConfig.HEADER_CONVERTER_VERSION_CONFIG));
    }

    private void mockVersionedTaskHeaderConverterFromWorker(HeaderConverter returning) {
        when(plugins.newHeaderConverter(any(WorkerConfig.class), eq(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(WorkerConfig.HEADER_CONVERTER_VERSION)))
            .thenReturn(returning);
    }

    private void verifyVersionedTaskHeaderConverterFromWorker() {
        verify(plugins).newHeaderConverter(any(WorkerConfig.class), eq(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(WorkerConfig.HEADER_CONVERTER_VERSION));
    }

    private void mockGenericIsolation(boolean connector) {
        if (!connector) {
            when(plugins.withClassLoader(null)).thenReturn(loaderSwap);
        } else {
            when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);
        }
    }

    private void verifyGenericIsolation() {
        verify(loaderSwap, atLeastOnce()).close();
    }

    private void mockConnectorIsolation(String connectorClass, Connector connector) {
        mockGenericIsolation(true);
        when(plugins.newConnector(connectorClass)).thenReturn(connector);
        when(connector.version()).thenReturn("1.0");
    }

    private void mockVersionedConnectorIsolation(String connectorClass, VersionRange range, Connector connector) {
        mockGenericIsolation(true);
        when(plugins.newConnector(connectorClass, range)).thenReturn(connector);
        when(connector.version()).thenReturn(range == null ? "unknown" : range.toString());
    }

    private void verifyConnectorIsolation(Connector connector) {
        verifyGenericIsolation();
        verify(plugins).newConnector(anyString());
        verify(connector, atLeastOnce()).version();

Review comment (bug):
nit: If we're calling withClassLoader with a null value, it means that some other call is returning a null ClassLoader, which is suspicious of a bug or incomplete mocking.

Code after:
        verify(plugins).newHeaderConverter(any(ConnectorConfig.class), eq(ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(ConnectorConfig.HEADER_CONVERTER_VERSION_CONFIG));
    }

    private void mockVersionedTaskHeaderConverterFromWorker(HeaderConverter returning) {
        when(plugins.newHeaderConverter(any(WorkerConfig.class), eq(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(WorkerConfig.HEADER_CONVERTER_VERSION)))
            .thenReturn(returning);
    }

    private void verifyVersionedTaskHeaderConverterFromWorker() {
        verify(plugins).newHeaderConverter(any(WorkerConfig.class), eq(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG), eq(WorkerConfig.HEADER_CONVERTER_VERSION));
    }

    private void mockGenericIsolation() {
        when(plugins.connectorLoader(any(), any())).thenReturn(pluginLoader);
        when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);
    }

    private void verifyGenericIsolation() {
        verify(loaderSwap, atLeastOnce()).close();
    }

    private void mockConnectorIsolation(String connectorClass, Connector connector) {
        mockGenericIsolation();
        when(plugins.newConnector(connectorClass)).thenReturn(connector);
        when(connector.version()).thenReturn("1.0");
    }

    private void mockVersionedConnectorIsolation(String connectorClass, VersionRange range, Connector connector) {
        mockGenericIsolation();
        when(plugins.newConnector(connectorClass, range)).thenReturn(connector);
        when(connector.version()).thenReturn(range == null ? "unknown" : range.toString());
    }

    private void verifyConnectorIsolation(Connector connector) {
        verifyGenericIsolation();
        verify(plugins).newConnector(anyString());
        verify(connector, atLeastOnce()).version();
    }

    private void verifyVersionedConnectorIsolation(String connectorClass, VersionRange range, Connector connector) {
        verifyGenericIsolation();
        verify(plugins).newConnector(connectorClass, range);
        verify(connector, atLeastOnce()).version();
    }

    private void mockTaskIsolation(Class<? extends Connector> connector, Class<? extends Task> taskClass, Task task) {
        mockGenericIsolation();
        doReturn(connector).when(plugins).connectorClass(connector.getName());
        when(plugins.newTask(taskClass)).thenReturn(task);
        when(task.version()).thenReturn("1.0");
    }