@Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> options = new HashSet<>();
        options.add(MySqlSourceOptions.HOSTNAME);
        options.add(MySqlSourceOptions.USERNAME);
        options.add(MySqlSourceOptions.PASSWORD);
        options.add(MySqlSourceOptions.DATABASE_NAME);
        options.add(MySqlSourceOptions.TABLE_NAME);
        return options;
    }
@Test
    public void testValidation() {
        // validate illegal port
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("port", "123b");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertTrue(
                    ExceptionUtils.findThrowableWithMessage(
                                    t, "Could not parse value '123b' for key 'port'.")
                            .isPresent());
        }

        // validate illegal server id
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("server-id", "123b");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertTrue(
                    ExceptionUtils.findThrowableWithMessage(
                                    t, "The value of option 'server-id' is invalid: '123b'")
                            .isPresent());
            assertTrue(
                    ExceptionUtils.findThrowableWithMessage(
                                    t, "The server id 123b is not a valid numeric.")
                            .isPresent());
        }

        // validate illegal connect.timeout
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("connect.timeout", "240ms");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertTrue(
                    ExceptionUtils.findThrowableWithMessage(
                                    t,
                                    "The value of option 'connect.timeout' cannot be less than PT0.25S, but actual is PT0.24S")
                            .isPresent());
        }

        // validate illegal split size
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("scan.incremental.snapshot.chunk.size", "1");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertThat(
                    t,
                    containsMessage(
                            "The value of option 'scan.incremental.snapshot.chunk.size' must larger than 1, but is 1"));
        }

        // validate illegal fetch size
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("scan.snapshot.fetch.size", "1");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertThat(
                    t,
                    containsMessage(
                            "The value of option 'scan.snapshot.fetch.size' must larger than 1, but is 1"));
        }

        // validate illegal split meta group size
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("chunk-meta.group.size", "1");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertThat(
                    t,
                    containsMessage(
                            "The value of option 'chunk-meta.group.size' must larger than 1, but is 1"));
        }

        // validate illegal split meta group size
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("split-key.even-distribution.factor.upper-bound", "0.8");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertThat(
                    t,
                    containsMessage(
                            "The value of option 'chunk-key.even-distribution.factor.upper-bound' must larger than or equals 1.0, but is 0.8"));
        }

        // validate illegal connection pool size
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("connection.pool.size", "1");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertThat(
                    t,
                    containsMessage(
                            "The value of option 'connection.pool.size' must larger than 1, but is 1"));
        }

        // validate illegal connect max retry times
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.incremental.snapshot.enabled", "true");
            properties.put("connect.max-retries", "0");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertThat(
                    t,
                    containsMessage(
                            "The value of option 'connect.max-retries' must larger than 0, but is 0"));
        }

        // validate missing required
        Factory factory = new MySqlTableSourceFactory();
        for (ConfigOption<?> requiredOption : factory.requiredOptions()) {
            Map<String, String> properties = getAllOptions();
            properties.remove(requiredOption.key());

            try {
                createTableSource(properties);
                fail("exception expected");
            } catch (Throwable t) {
                assertTrue(
                        ExceptionUtils.findThrowableWithMessage(
                                        t,
                                        "Missing required options are:\n\n" + requiredOption.key())
                                .isPresent());
            }
        }

        // validate unsupported option
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("unknown", "abc");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            assertTrue(
                    ExceptionUtils.findThrowableWithMessage(t, "Unsupported options:\n\nunknown")
                            .isPresent());
        }

        // validate unsupported option
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("scan.startup.mode", "abc");

            createTableSource(properties);
            fail("exception expected");
        } catch (Throwable t) {
            String msg =
                    "Invalid value for option 'scan.startup.mode'. Supported values are "
                            + "[initial, snapshot, latest-offset, earliest-offset, specific-offset, timestamp], "
                            + "but was: abc";
            assertTrue(ExceptionUtils.findThrowableWithMessage(t, msg).isPresent());
        }

        // validate invalid database-name
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("database-name", "*_invalid_db");
        } catch (Throwable t) {
            String msg =
                    String.format(
                            "The database-name '%s' is not a valid regular expression",
                            "*_invalid_db");
            assertTrue(ExceptionUtils.findThrowableWithMessage(t, msg).isPresent());
        }
        // validate invalid table-name
        try {
            Map<String, String> properties = getAllOptions();
            properties.put("table-name", "*_invalid_table");
        } catch (Throwable t) {
            String msg =
                    String.format(
                            "The table-name '%s' is not a valid regular expression",
                            "*_invalid_table");
            assertTrue(ExceptionUtils.findThrowableWithMessage(t, msg).isPresent());
        }
    }