### Question:
LocalFileSystemScanner implements FileSystemScanner { @Override public void scan(final ConnectorContext context) { cleanUpCompletedFiles(); if (updateFiles()) { LOG.info("Requesting task reconfiguration"); context.requestTaskReconfiguration(); } } LocalFileSystemScanner(final String sourceDirectoryPath,
                                  final FSDirectoryWalker fsWalker,
                                  final GenericFileCleanupPolicy cleaner,
                                  final OffsetManager offsetManager,
                                  final StateBackingStore<SourceFile> store); @Override void scan(final ConnectorContext context); @Override synchronized List<List<String>> partitionFilesAndGet(int maxGroups); @Override void close();  }### Answer:
@Test public void shouldCleanRemainingCompletedFilesWhileStarting() { Map<String, SourceFile> states = new HashMap<String, SourceFile>() { { put("???", INPUT_FILES.stateFor(0, SourceStatus.COMPLETED)); } }; StateSnapshot<SourceFile> state = new StateSnapshot<>(0, states); FileStateBackingStore store = Mockito.mock(FileStateBackingStore.class); Mockito.when(store.snapshot()).thenReturn(state); final List<File> sources = INPUT_FILES.getInputPathsFor(0, 1); final MockTimesDirectoryScanner ds = new MockTimesDirectoryScanner(sources); final MockFileCleaner cleaner = new MockFileCleaner(true); LocalFileSystemScanner scanner = newFsMonitorThread(cleaner, ds, store); scanner.scan(new MockConnectorContext()); assertEquals(1, cleaner.getSucceed().size()); assertEquals(sources.get(0), cleaner.getSucceed().get(0)); }
@Test public void shouldNotScannedDirectoryWhileProcessingFiles() { final StateSnapshot<SourceFile> state = new StateSnapshot<>(0, Collections.emptyMap()); FileStateBackingStore store = Mockito.mock(FileStateBackingStore.class); Mockito.when(store.snapshot()).thenReturn(state); final List<File> sources = INPUT_FILES.getInputPathsFor(0, 1); final MockTimesDirectoryScanner ds = new MockTimesDirectoryScanner(sources); final LocalFileSystemScanner scanner = newFsMonitorThread(new MockFileCleaner(true), ds, store); MockConnectorContext context = new MockConnectorContext(); scanner.scan(context); scanner.scan(context); assertEquals(1, ds.times()); }
@Test public void shouldCleanUpFilesAfterReceivingCompletedState() { final StateSnapshot<SourceFile> state = new StateSnapshot<>(0, Collections.emptyMap()); final InMemoryStateBackingStore<SourceFile> store = new InMemoryStateBackingStore<>(state); final MockFileCleaner cleaner = new MockFileCleaner(true); final MockTimesDirectoryScanner ds = new MockTimesDirectoryScanner(); final LocalFileSystemScanner scanner = newFsMonitorThread(cleaner, ds, store); ds.put(INPUT_FILES.getInputPathsFor(0, 1)); scanner.scan(new MockConnectorContext()); assertEquals(0, cleaner.getSucceed().size()); assertEquals(0, cleaner.getFailed().size()); final SourceFile completed = INPUT_FILES.stateFor(0, SourceStatus.COMPLETED); store.listener.onStateUpdate(OFFSET_MANAGER.toPartitionJson(completed.metadata()), completed); final SourceFile failed = INPUT_FILES.stateFor(1, SourceStatus.FAILED); store.listener.onStateUpdate(OFFSET_MANAGER.toPartitionJson(failed.metadata()), failed); ds.put(INPUT_FILES.getInputPathsFor(0, 1, 2, 3)); scanner.scan(new MockConnectorContext()); assertEquals(2, ds.times()); assertEquals(1, cleaner.getSucceed().size()); assertEquals(INPUT_FILES.metadataFor(0).absolutePath(), cleaner.getSucceed().get(0).getAbsolutePath()); assertEquals(1, cleaner.getFailed().size()); assertEquals(INPUT_FILES.metadataFor(1).absolutePath(), cleaner.getFailed().get(0).getAbsolutePath()); }