List<TaskDirectory> listAllTaskDirectories() {
        return listTaskDirectories(pathname -> pathname.isDirectory() && TASK_DIR_PATH_NAME.matcher(pathname.getName()).matches());
    }
@Test
    public void shouldReturnEmptyArrayIfListFilesReturnsNull() throws IOException {
        stateDir = new File(TestUtils.IO_TMP_DIR, "kafka-" + TestUtils.randomString(5));
        directory = new StateDirectory(
            new StreamsConfig(new Properties() {
                {
                    put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
                    put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
                    put(StreamsConfig.STATE_DIR_CONFIG, stateDir.getPath());
                }
            }),
            time,
            true,
            false);
        appDir = new File(stateDir, applicationId);

        // make sure the File#listFiles returns null and StateDirectory#listAllTaskDirectories is able to handle null
        Utils.delete(appDir);
        Files.createFile(appDir.toPath());
        assertTrue(Files.exists(appDir.toPath()));
        assertNull(appDir.listFiles());
        assertEquals(0, directory.listAllTaskDirectories().size());
    }