Postgres cdc: check whether replication slot name is duplicate on upstream about risingwave

[StrikeW]

We already check whether the slot.name has already existed for postgres-cdc.

risingwave/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/common/PostgresValidator.java

Lines 96 to 115 in 293db73

	try (var stmt =
	jdbcConnection.prepareStatement(ValidatorUtils.getSql("postgres.slot.check"))) {
	// check whether the replication slot is already existed
	stmt.setString(1, slotName);
	stmt.setString(2, dbName);
	var res = stmt.executeQuery();
	if (res.next() && res.getString(1).equals(slotName)) {
	LOG.info("replication slot '{}' already exists, just use it", slotName);
	} else {
	// otherwise, we need to create a new one
	var stmt2 =
	jdbcConnection.prepareStatement(
	ValidatorUtils.getSql("postgres.slot_limit.check"));
	var res2 = stmt2.executeQuery();
	// check whether the number of replication slots reaches the max limit
	if (res2.next() && res2.getString(1).equals("true")) {
	throw ValidatorUtils.failedPrecondition(
	"all replication slots are in use\n Hint: Free one or increase max_replication_slots.");
	}
	}

But there is a race condition:
If user creates two postgres sources concurrently with the same slot.name, both sources can pass the validation because the slot doesn't exist indeed. But one of the source job will fail when the source executor start the Debezium connector, due to the duplication of slot.name.

Options:

• One solution is to limit the concurrency of CREATE SOURCE DDL to 1 to prevent creating multiple PG sources concurrently
• Maintain a internal cache/table on Meta to do the duplication check