public static List<String> parseColumnsFromPath(String filePath, List<String> columnsFromPath)
            throws UserException {
        if (columnsFromPath == null || columnsFromPath.isEmpty()) {
            return Collections.emptyList();
        }
        String[] strings = filePath.split("/");
        if (strings.length < 2) {
            throw new UserException(
                    "Fail to parse columnsFromPath, expected: " + columnsFromPath + ", filePath: " + filePath);
        }
        String[] columns = new String[columnsFromPath.size()];
        int size = 0;
        for (int i = strings.length - 2; i >= 0; i--) {
            String str = strings[i];
            if (str != null && str.isEmpty()) {
                continue;
            }
            if (str == null || !str.contains("=")) {
                throw new UserException(
                        "Fail to parse columnsFromPath, expected: " + columnsFromPath + ", filePath: " + filePath);
            }
            String[] pair = str.split("=", 2);
            if (pair.length != 2) {
                throw new UserException(
                        "Fail to parse columnsFromPath, expected: " + columnsFromPath + ", filePath: " + filePath);
            }
            int index = columnsFromPath.indexOf(pair[0]);
            if (index == -1) {
                continue;
            }
            columns[index] = pair[1];
            size++;
            if (size >= columnsFromPath.size()) {
                break;
            }
        }
        if (size != columnsFromPath.size()) {
            throw new UserException(
                    "Fail to parse columnsFromPath, expected: " + columnsFromPath + ", filePath: " + filePath);
        }
        return Lists.newArrayList(columns);
    }
@Test
    public void parseColumnsFromPath() {
        String path = "/path/to/dir/k1=v1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Collections.singletonList("k1"));
            assertEquals(1, columns.size());
            assertEquals(Collections.singletonList("v1"), columns);
        } catch (UserException e) {
            fail();
        }

        path = "/path/to/dir/k1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Collections.singletonList("k1"));
            fail();
        } catch (UserException ignored) {
        }

        path = "/path/to/dir/k1=v1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Collections.singletonList("k2"));
            fail();
        } catch (UserException ignored) {
        }

        path = "/path/to/dir/k1=v2/k1=v1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Collections.singletonList("k1"));
            assertEquals(1, columns.size());
            assertEquals(Collections.singletonList("v1"), columns);
        } catch (UserException e) {
            fail();
        }

        path = "/path/to/dir/k2=v2/k1=v1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Lists.newArrayList("k1", "k2"));
            assertEquals(2, columns.size());
            assertEquals(Lists.newArrayList("v1", "v2"), columns);
        } catch (UserException e) {
            fail();
        }

        path = "/path/to/dir/k2=v2/a/k1=v1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Lists.newArrayList("k1", "k2"));
            fail();
        } catch (UserException ignored) {
        }

        path = "/path/to/dir/k2=v2/k1=v1/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Lists.newArrayList("k1", "k2", "k3"));
            fail();
        } catch (UserException ignored) {
        }

        path = "/path/to/dir/k2=v2//k1=v1//xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Lists.newArrayList("k1", "k2"));
            assertEquals(2, columns.size());
            assertEquals(Lists.newArrayList("v1", "v2"), columns);
        } catch (UserException e) {
            fail();
        }

        path = "/path/to/dir/k2==v2=//k1=v1//xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Lists.newArrayList("k1", "k2"));
            assertEquals(2, columns.size());
            assertEquals(Lists.newArrayList("v1", "=v2="), columns);
        } catch (UserException e) {
            fail();
        }

        path = "/path/to/dir/k2==v2=//k1=v1/";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Lists.newArrayList("k1", "k2"));
            fail();
        } catch (UserException ignored) {
        }

        path = "/path/to/dir/k1=2/a/xxx.csv";
        try {
            List<String> columns = BrokerUtil.parseColumnsFromPath(path, Collections.singletonList("k1"));
            fail();
        } catch (UserException ignored) {
            ignored.printStackTrace();
        }

    }