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

Code before:
        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 testClose() throws Exception {
        long timeoutMs = 1000;

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

        committer.close(timeoutMs);

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

        committer.close(timeoutMs);

        verify(executor, times(2)).shutdown();
    }

    @Test
    public void testRemove() throws Exception {
        // Try to remove a non-existing task
        assertTrue(committers.isEmpty());
        committer.remove(taskId);
        assertTrue(committers.isEmpty());

        // Try to remove an existing task
        when(taskFuture.cancel(false)).thenReturn(false);
        when(taskFuture.isDone()).thenReturn(false);
        when(taskFuture.get())
                .thenReturn(null)
                .thenThrow(new CancellationException())
                .thenThrow(new InterruptedException());
        when(taskId.connector()).thenReturn("MyConnector");
        when(taskId.task()).thenReturn(1);

        committers.put(taskId, taskFuture);
        committer.remove(taskId);
        assertTrue(committers.isEmpty());

        // Try to remove a cancelled task

Review comment (suggestion):
We want to verify here that at least one `ERROR`-level log message was emitted by the `SourceTaskOffsetCommitter` instance during this part of the test.

We can use the `LogCaptureAppender` class to do that exact thing, without installing a mock logger for the class:
```java
        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            committer.close(timeoutMs);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("ERROR")));
        }
```

Code after:
    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();
    }

    @Test
    public void testRemoveNonExistentTask() {
        assertTrue(committers.isEmpty());
        committer.remove(taskId);
        assertTrue(committers.isEmpty());
    }

    @Test
    public void testRemoveSuccess() {
        expectRemove();
        committers.put(taskId, taskFuture);