@Override
    public boolean acceptsURL(String url) throws SQLException {
        return DriverUri.acceptsURL(url);
    }
@Test
    public void testDriverInvalidUri() throws Exception {
        // Invalid prefix
        for (String invalidPrefixUri :
                Arrays.asList(
                        "flink://localhost:8888/catalog_name/database_name?sessionId=123&key1=val1&key2=val2",
                        "jdbc::flink://localhost:8888/catalog_name/database_name?sessionId=123&key1=val1&key2=val2",
                        "jdbc::flink//localhost:8888/catalog_name/database_name?sessionId=123&key1=val1&key2=val2")) {
            assertFalse(DriverUri.acceptsURL(invalidPrefixUri));
            assertThrowsExactly(
                    SQLException.class,
                    () -> DriverUri.create(invalidPrefixUri, new Properties()),
                    String.format(
                            "Flink JDBC URL[%s] must start with [jdbc:flink:]", invalidPrefixUri));
        }

        // Without host or port
        String noPortUri = "jdbc:flink://localhost/catalog";
        assertThrowsExactly(
                SQLException.class,
                () -> DriverUri.create(noPortUri, new Properties()),
                String.format("No port specified in uri: %s", noPortUri));

        Properties properties = new Properties();
        properties.setProperty("key3", "val33");
        for (String dupPropUri :
                Arrays.asList(
                        "jdbc:flink://localhost:8088/catalog?key1=val1&key2=val2&key3=val3",
                        "jdbc:flink://localhost:8088/catalog?key1=val1&key2=val2&key1=val3")) {
            assertThrowsExactly(
                    SQLException.class,
                    () -> DriverUri.create(dupPropUri, properties),
                    "Connection property 'key3' is both in the URL and an argument");
        }
    }