@Nonnull
    public static BatchSource<String> files(@Nonnull String directory) {
        return filesBuilder(directory).build();
    }
@Test
    public void files() throws Exception {
        // Given
        File directory = createTempDirectory();
        File file1 = new File(directory, randomName());
        appendToFile(file1, "hello", "world");
        File file2 = new File(directory, randomName());
        appendToFile(file2, "hello2", "world2");

        // When
        BatchSource<String> source = Sources.files(directory.getPath());

        // Then
        p.readFrom(source).writeTo(sink);
        execute();
        int nodeCount = hz().getCluster().getMembers().size();
        assertEquals(4 * nodeCount, sinkList.size());
    }