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

Code before:
        workerProps.put("key.converter", "org.apache.kafka.connect.json.JsonConverter");
        workerProps.put("value.converter", "org.apache.kafka.connect.json.JsonConverter");
        workerProps.put("offset.storage.file.filename", "/tmp/connect.offsets");
        workerProps.put("offset.flush.interval.ms",
                Long.toString(DEFAULT_OFFSET_COMMIT_INTERVAL_MS));
        WorkerConfig config = new StandaloneConfig(workerProps);
        committer = new SourceTaskOffsetCommitter(config, executor, committers);
    }

    @SuppressWarnings("unchecked")
    @Test
    public void testSchedule() {
        ArgumentCaptor<Runnable> taskWrapper = ArgumentCaptor.forClass(Runnable.class);

        when(executor.scheduleWithFixedDelay(
                taskWrapper.capture(), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
                eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS), eq(TimeUnit.MILLISECONDS))
        ).thenReturn((ScheduledFuture) commitFuture);

        committer.schedule(taskId, task);
        assertNotNull(taskWrapper.getValue());
        assertEquals(singletonMap(taskId, commitFuture), committers);
    }

    @Test
    public void testCloseWithinTimeout() throws Exception {
        long timeoutMs = 1000;

        // Normal termination, where termination times out.
        when(executor.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)).thenReturn(false);

        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            committer.close(timeoutMs);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("ERROR")));
        }

        verify(executor).shutdown();
    }

    @Test
    public void testCloseOutsideOfTimeout() throws InterruptedException {
        long timeoutMs = 1000;

        // Termination interrupted
        when(executor.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)).thenThrow(new InterruptedException());

        committer.close(timeoutMs);

        verify(executor).shutdown();
    }


Review comment (bug):
Nit: this name seems to go against the case that we're covering here (where `close` fails to complete within the expected timeout). Maybe rename to `testCloseTimeout`?

Code after:
        workerProps.put("key.converter", "org.apache.kafka.connect.json.JsonConverter");
        workerProps.put("value.converter", "org.apache.kafka.connect.json.JsonConverter");
        workerProps.put("offset.storage.file.filename", "/tmp/connect.offsets");
        workerProps.put("offset.flush.interval.ms",
                Long.toString(DEFAULT_OFFSET_COMMIT_INTERVAL_MS));
        WorkerConfig config = new StandaloneConfig(workerProps);
        committer = new SourceTaskOffsetCommitter(config, executor, committers);
    }

    @SuppressWarnings("unchecked")
    @Test
    public void testSchedule() {
        ArgumentCaptor<Runnable> taskWrapper = ArgumentCaptor.forClass(Runnable.class);

        when(executor.scheduleWithFixedDelay(
                taskWrapper.capture(), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
                eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS), eq(TimeUnit.MILLISECONDS))
        ).thenReturn((ScheduledFuture) commitFuture);

        committer.schedule(taskId, task);
        assertNotNull(taskWrapper.getValue());
        assertEquals(singletonMap(taskId, commitFuture), committers);
    }

    @Test
    public void testCloseTimeout() throws Exception {
        long timeoutMs = 1000;

        // Normal termination, where termination times out.
        when(executor.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)).thenReturn(false);

        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            committer.close(timeoutMs);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("ERROR")));
        }

        verify(executor).shutdown();
    }

    @Test
    public void testCloseInterrupted() throws InterruptedException {
        long timeoutMs = 1000;

        // Termination interrupted
        when(executor.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)).thenThrow(new InterruptedException());

        committer.close(timeoutMs);

        verify(executor).shutdown();
    }