feat(connector): support ingest Citus distributed table

java/connector-node/connector-api/src/main/java/com/risingwave/connector/api/source/SourceTypeE.java

@@ -19,6 +19,7 @@
 public enum SourceTypeE {
     MYSQL,
     POSTGRES,
+    CITUS,
     INVALID;
 
     public static SourceTypeE valueOf(ConnectorServiceProto.SourceType type) {
@@ -27,6 +28,8 @@ public static SourceTypeE valueOf(ConnectorServiceProto.SourceType type) {
                 return SourceTypeE.MYSQL;
             case POSTGRES:
                 return SourceTypeE.POSTGRES;
+            case CITUS:
+                return SourceTypeE.CITUS;
             default:
                 return SourceTypeE.INVALID;
         }

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/sourcenode/common/DbzConnectorConfig.java

@@ -36,13 +36,16 @@ public class DbzConnectorConfig {
     public static final String DB_NAME = "database.name";
     public static final String TABLE_NAME = "table.name";
 
+    public static final String DB_SERVERS = "database.servers";
+
     /* MySQL specified configs */
     public static final String MYSQL_SERVER_ID = "server.id";
 
     /* Postgres specified configs */
     public static final String PG_SLOT_NAME = "slot.name";
     public static final String PG_SCHEMA_NAME = "schema.name";
 
+    private static final String DBZ_CONFIG_FILE = "debezium.properties";
     private static final String MYSQL_CONFIG_FILE = "mysql.properties";
     private static final String POSTGRES_CONFIG_FILE = "postgres.properties";
 
@@ -81,15 +84,8 @@ public Properties getResolvedDebeziumProps() {
 
     public DbzConnectorConfig(
             SourceTypeE source, long sourceId, String startOffset, Map<String, String> userProps) {
-        var dbzProps = new Properties();
-        try (var input = getClass().getClassLoader().getResourceAsStream("debezium.properties")) {
-            assert input != null;
-            dbzProps.load(input);
-        } catch (IOException e) {
-            throw new RuntimeException("failed to load debezium.properties", e);
-        }
-
         StringSubstitutor substitutor = new StringSubstitutor(userProps);
+        var dbzProps = initiateDbConfig(DBZ_CONFIG_FILE, substitutor);
         if (source == SourceTypeE.MYSQL) {
             var mysqlProps = initiateDbConfig(MYSQL_CONFIG_FILE, substitutor);
             // if offset is specified, we will continue binlog reading from the specified offset
@@ -104,9 +100,14 @@ public DbzConnectorConfig(
             }
 
             dbzProps.putAll(mysqlProps);
-        } else if (source == SourceTypeE.POSTGRES) {
+        } else if (source == SourceTypeE.POSTGRES || source == SourceTypeE.CITUS) {
             var postgresProps = initiateDbConfig(POSTGRES_CONFIG_FILE, substitutor);
 
+            // citus needs all_tables publication to capture all shards
+            if (source == SourceTypeE.CITUS) {
+                postgresProps.setProperty("publication.autocreate.mode", "all_tables");
+            }
+
             // if offset is specified, we will continue reading changes from the specified offset
             if (null != startOffset && !startOffset.isBlank()) {
                 postgresProps.setProperty("snapshot.mode", "never");
@@ -134,7 +135,7 @@ private Properties initiateDbConfig(String fileName, StringSubstitutor substitut
             var resolvedStr = substitutor.replace(inputStr);
             dbProps.load(new StringReader(resolvedStr));
         } catch (IOException e) {
-            throw new RuntimeException("failed to load " + fileName, e);
+            throw new RuntimeException("failed to load config file " + fileName, e);
         }
         return dbProps;
     }