public void remove(ConnectorTaskId id) {
        final ScheduledFuture<?> task = committers.remove(id);
        if (task == null)
            return;

        try (LoggingContext loggingContext = LoggingContext.forTask(id)) {
            task.cancel(false);
            if (!task.isDone())
                task.get();
        } catch (CancellationException e) {
            // ignore
            log.trace("Offset commit thread was cancelled by another thread while removing connector task with id: {}", id);
        } catch (ExecutionException | InterruptedException e) {
            throw new ConnectException("Unexpected interruption in SourceTaskOffsetCommitter while removing task with id: " + id, e);
        }
    }
@Test
    public void testRemoveTaskAndInterrupted() throws ExecutionException, InterruptedException {
        expectRemove();
        when(taskFuture.get()).thenThrow(new InterruptedException());

        committers.put(taskId, taskFuture);
        assertThrows(ConnectException.class, () -> committer.remove(taskId));
    }