public DebeziumSourceFunction<T> build() {
            Properties props = new Properties();
            props.setProperty("connector.class", SqlServerConnector.class.getCanonicalName());
            // hard code server name, because we don't need to distinguish it, docs:
            // Logical name that identifies and provides a namespace for the SQL Server database
            // server that you want Debezium to capture. The logical name should be unique across
            // all other connectors, since it is used as a prefix for all Kafka topic names
            // emanating from this connector. Only alphanumeric characters and underscores should be
            // used.
            props.setProperty("database.server.name", DATABASE_SERVER_NAME);
            props.setProperty("database.hostname", checkNotNull(hostname));
            props.setProperty("database.user", checkNotNull(username));
            props.setProperty("database.password", checkNotNull(password));
            props.setProperty("database.port", String.valueOf(port));
            props.setProperty("database.history.skip.unparseable.ddl", String.valueOf(true));
            props.setProperty("database.dbname", checkNotNull(database));

            if (tableList != null) {
                props.setProperty("table.include.list", String.join(",", tableList));
            }

            switch (startupOptions.startupMode) {
                case INITIAL:
                    props.setProperty("snapshot.mode", "initial");
                    break;
                case INITIAL_ONLY:
                    props.setProperty("snapshot.mode", "initial_only");
                    break;
                case LATEST_OFFSET:
                    props.setProperty("snapshot.mode", "schema_only");
                    break;
                default:
                    throw new UnsupportedOperationException();
            }

            if (dbzProperties != null) {
                props.putAll(dbzProperties);
            }

            return new DebeziumSourceFunction<>(
                    deserializer, props, null, new SqlServerValidator(props), metricOption);
        }
metricOption used to instantiate SourceExactlyMetric when inlong.metric.labels is present in flink sql
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/sqlserver/SqlServerSource.java