@Test public void testSchedule() throws Exception { Capture<Runnable> taskWrapper = EasyMock.newCapture(); ScheduledFuture commitFuture = PowerMock.createMock(ScheduledFuture.class); EasyMock.expect(executor.scheduleWithFixedDelay( EasyMock.capture(taskWrapper), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS), eq(TimeUnit.MILLISECONDS)) ).andReturn(commitFuture); ConnectorTaskId taskId = PowerMock.createMock(ConnectorTaskId.class); WorkerSourceTask task = PowerMock.createMock(WorkerSourceTask.class); EasyMock.expect(committers.put(taskId, commitFuture)).andReturn(null); PowerMock.replayAll(); committer.schedule(taskId, task); assertTrue(taskWrapper.hasCaptured()); assertNotNull(taskWrapper.getValue()); PowerMock.verifyAll(); }
public void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) { long commitIntervalMs = config.getLong(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG); ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(new Runnable() { @Override public void run() { commit(workerTask); } }, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS); committers.put(id, commitFuture); }
SourceTaskOffsetCommitter { public void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) { long commitIntervalMs = config.getLong(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG); ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(new Runnable() { @Override public void run() { commit(workerTask); } }, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS); committers.put(id, commitFuture); } }
SourceTaskOffsetCommitter { public void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) { long commitIntervalMs = config.getLong(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG); ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(new Runnable() { @Override public void run() { commit(workerTask); } }, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS); committers.put(id, commitFuture); } SourceTaskOffsetCommitter(WorkerConfig config,
                              ScheduledExecutorService commitExecutorService,
                              ConcurrentMap<ConnectorTaskId, ScheduledFuture<?>> committers);  SourceTaskOffsetCommitter(WorkerConfig config); }
SourceTaskOffsetCommitter { public void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) { long commitIntervalMs = config.getLong(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG); ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(new Runnable() { @Override public void run() { commit(workerTask); } }, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS); committers.put(id, commitFuture); } 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 schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) { long commitIntervalMs = config.getLong(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG); ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(new Runnable() { @Override public void run() { commit(workerTask); } }, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS); committers.put(id, commitFuture); } 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);  }