@Nonnull
    public static StreamSource<String> fileWatcher(@Nonnull String watchedDirectory) {
        return filesBuilder(watchedDirectory).buildWatcher();
    }
@Test
    @Ignore("Changes on the file is not reflected as an event from the File System, needs more investigation")
    public void fileChanges() throws Exception {
        // Given
        File directory = createTempDirectory();
        // this is a pre-existing file, should not be picked up
        File file = new File(directory, randomName());
        appendToFile(file, "hello", "pre-existing");
        sleepAtLeastMillis(50);

        // When
        StreamSource<String> source = Sources.fileWatcher(directory.getPath());

        // Then
        p.readFrom(source).withoutTimestamps().writeTo(sink);
        Job job = hz().getJet().newJob(p);
        // wait for the processor to initialize
        assertThat(job).eventuallyHasStatus(JobStatus.RUNNING);
        // pre-existing file should not be picked up
        assertEquals(0, sinkList.size());
        appendToFile(file, "third line");
        // now, only new line should be picked up
        int nodeCount = hz().getCluster().getMembers().size();
        assertTrueEventually(() -> assertEquals(nodeCount, sinkList.size()));
    }