@Test public void testRemove() throws Exception { ConnectorTaskId taskId = PowerMock.createMock(ConnectorTaskId.class); ScheduledFuture task = PowerMock.createMock(ScheduledFuture.class); EasyMock.expect(committers.remove(taskId)).andReturn(null); PowerMock.replayAll(); committer.remove(taskId); PowerMock.verifyAll(); PowerMock.resetAll(); EasyMock.expect(committers.remove(taskId)).andReturn(task); EasyMock.expect(task.cancel(eq(false))).andReturn(false); EasyMock.expect(task.isDone()).andReturn(false); EasyMock.expect(task.get()).andReturn(null); PowerMock.replayAll(); committer.remove(taskId); PowerMock.verifyAll(); PowerMock.resetAll(); EasyMock.expect(committers.remove(taskId)).andReturn(task); EasyMock.expect(task.cancel(eq(false))).andReturn(false); EasyMock.expect(task.isDone()).andReturn(false); EasyMock.expect(task.get()).andThrow(new CancellationException()); mockLog.trace(EasyMock.anyString(), EasyMock.<Object>anyObject()); PowerMock.expectLastCall(); PowerMock.replayAll(); committer.remove(taskId); PowerMock.verifyAll(); PowerMock.resetAll(); EasyMock.expect(committers.remove(taskId)).andReturn(task); EasyMock.expect(task.cancel(eq(false))).andReturn(false); EasyMock.expect(task.isDone()).andReturn(false); EasyMock.expect(task.get()).andThrow(new InterruptedException()); PowerMock.replayAll(); try { committer.remove(taskId); fail("Expected ConnectException to be raised"); } catch (ConnectException e) { } PowerMock.verifyAll(); }
public void remove(ConnectorTaskId id) { final ScheduledFuture<?> task = committers.remove(id); if (task == null) return; try { task.cancel(false); if (!task.isDone()) task.get(); } catch (CancellationException e) { 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); } }
SourceTaskOffsetCommitter { public void remove(ConnectorTaskId id) { final ScheduledFuture<?> task = committers.remove(id); if (task == null) return; try { task.cancel(false); if (!task.isDone()) task.get(); } catch (CancellationException e) { 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); } } }
SourceTaskOffsetCommitter { public void remove(ConnectorTaskId id) { final ScheduledFuture<?> task = committers.remove(id); if (task == null) return; try { task.cancel(false); if (!task.isDone()) task.get(); } catch (CancellationException e) { 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); } } SourceTaskOffsetCommitter(WorkerConfig config,
                              ScheduledExecutorService commitExecutorService,
                              ConcurrentMap<ConnectorTaskId, ScheduledFuture<?>> committers);  SourceTaskOffsetCommitter(WorkerConfig config); }
SourceTaskOffsetCommitter { public void remove(ConnectorTaskId id) { final ScheduledFuture<?> task = committers.remove(id); if (task == null) return; try { task.cancel(false); if (!task.isDone()) task.get(); } catch (CancellationException e) { 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); } } SourceTaskOffsetCommitter(WorkerConfig config,
                              ScheduledExecutorService commitExecutorService,
                              ConcurrentMap<ConnectorTaskId, ScheduledFuture<?>> committers);  SourceTaskOffsetCommitter(WorkerConfig config); void close(long timeoutMs); void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask); void remove(ConnectorTaskId id); }
SourceTaskOffsetCommitter { public void remove(ConnectorTaskId id) { final ScheduledFuture<?> task = committers.remove(id); if (task == null) return; try { task.cancel(false); if (!task.isDone()) task.get(); } catch (CancellationException e) { 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); } } SourceTaskOffsetCommitter(WorkerConfig config,
                              ScheduledExecutorService commitExecutorService,
                              ConcurrentMap<ConnectorTaskId, ScheduledFuture<?>> committers);  SourceTaskOffsetCommitter(WorkerConfig config); void close(long timeoutMs); void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask); void remove(ConnectorTaskId id);  }