public static ConnectionString buildConnectionString(
            @Nullable String username,
            @Nullable String password,
            String hosts,
            @Nullable String connectionOptions) {
        StringBuilder sb = new StringBuilder(MONGODB_SCHEME).append("://");

        if (StringUtils.isNotEmpty(username) && StringUtils.isNotEmpty(password)) {
            sb.append(encodeValue(username)).append(":").append(encodeValue(password)).append("@");
        }

        sb.append(checkNotNull(hosts));

        if (StringUtils.isNotEmpty(connectionOptions)) {
            sb.append("/?").append(connectionOptions);
        }

        return new ConnectionString(sb.toString());
    }
Utilities of MongoDB operations.
Copy from com.ververica:flink-connector-mongodb-cdc:2.3.0.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/utils/MongoUtils.java