[INLONG-5046][Agent] Support collect data from PostgreSQL

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/PostgresSnapshotModeConstants.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.agent.constant;
+
+/**
+ * Constants of job fetcher Postgressnapshot mode
+ */
+public class PostgresSnapshotModeConstants {
+    public static final String INITIAL = "initial";
+
+    public static final String EARLIEST_OFFSET = "never";
+
+    public static final String ALWAYS = "always";
+
+    public static final String EXPORTED = "exported";
+
+    public static final String INITIAL_ONLY = "initial_only";
+
+    public static final String CUSTOM = "custom";
+}

inlong-agent/agent-plugins/pom.xml

@@ -88,6 +88,10 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-postgres</artifactId>
+        </dependency>
         <dependency>
             <artifactId>awaitility</artifactId>
             <groupId>org.awaitility</groupId>

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/PostgreSQLSource.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.agent.plugin.sources;
+
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.plugin.Reader;
+import org.apache.inlong.agent.plugin.Source;
+import org.apache.inlong.agent.plugin.sources.reader.PostgreSqlReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_STREAM_ID;
+import static org.apache.inlong.agent.constant.AgentConstants.GLOBAL_METRICS;
+
+/**
+ * PostgreSql source, split PostgreSql source job into multi readers
+ */
+public class PostgreSQLSource implements Source {
+
+    private static final Logger logger = LoggerFactory.getLogger(PostgreSQLSource.class);
+
+    private static final String POSTGRESQL_SOURCE_TAG_NAME = "PostgreSqlSourceMetric";
+
+    public PostgreSQLSource() {
+
+    }
+
+    @Override
+    public List<Reader> split(JobProfile conf) {
+        String inlongGroupId = conf.get(PROXY_INLONG_GROUP_ID, DEFAULT_PROXY_INLONG_GROUP_ID);
+        String inlongStreamId = conf.get(PROXY_INLONG_STREAM_ID, DEFAULT_PROXY_INLONG_STREAM_ID);
+        String metricTagName = POSTGRESQL_SOURCE_TAG_NAME + "_" + inlongGroupId + "_" + inlongStreamId;
+        Reader postgreReader = new PostgreSqlReader();
+        List<Reader> readerList = new ArrayList<>();
+        readerList.add(postgreReader);
+        GLOBAL_METRICS.incSourceSuccessCount(metricTagName);
+        return readerList;
+    }
+}