[INLONG-10318][Agent] Add PostgreSQL data source for Agent

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

@@ -97,6 +97,7 @@ public class TaskConstants extends CommonConstants {
  public static final String TASK_PULSAR_SUBSCRIPTION_POSITION = "task.pulsarTask.subscriptionPosition";
  public static final String TASK_PULSAR_RESET_TIME = "task.pulsarTask.resetTime";
 
+ // Mongo task
  public static final String TASK_MONGO_HOSTS = "task.mongoTask.hosts";
  public static final String TASK_MONGO_USER = "task.mongoTask.user";
  public static final String TASK_MONGO_PASSWORD = "task.mongoTask.password";
@@ -126,6 +127,18 @@ public class TaskConstants extends CommonConstants {
  public static final String TASK_MONGO_SSL_ENABLE = "task.mongoTask.sslEnabled";
  public static final String TASK_MONGO_POLL_INTERVAL = "task.mongoTask.pollIntervalInMs";
 
+ // PostgreSQL task
+ public static final String TASK_POSTGRES_HOSTNAME = "task.postgreSQLTask.hostname";
+ public static final String TASK_POSTGRES_PORT = "task.postgreSQLTask.port";
+ public static final String TASK_POSTGRES_USER = "task.postgreSQLTask.user";
+ public static final String TASK_POSTGRES_PASSWORD = "task.postgreSQLTask.password";
+ public static final String TASK_POSTGRES_DBNAME = "task.postgreSQLTask.dbname";
+ public static final String TASK_POSTGRES_SERVERNAME = "task.postgreSQLTask.servername";
+ public static final String TASK_POSTGRES_SCHEMA_INCLUDE_LIST = "task.postgreSQLTask.schemaIncludeList";
+ public static final String TASK_POSTGRES_TABLE_INCLUDE_LIST = "task.postgreSQLTask.tableIncludeList";
+ public static final String TASK_POSTGRES_PLUGIN_NAME = "task.postgreSQLTask.pluginName";
+ public static final String TASK_POSTGRES_SNAPSHOT_MODE = "task.postgreSQLTask.snapshotMode";
+
  public static final String TASK_STATE = "task.state";
 
  public static final String INSTANCE_STATE = "instance.state";

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/PostgreSQLTask.java

@@ -32,12 +32,11 @@ public class PostgreSQLTask {
  private String hostname;
  private String port;
  private String dbname;
- private String schema;
- private String servername;
- private String pluginname;
- private List<String> tableNameList;
+ private String schemaIncludeList;
+ private String pluginName;
+ private String tableIncludeList;
  private String serverTimeZone;
- private String scanStartupMode;
+ private String snapshotMode;
  private String primaryKey;
 
  @Data

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/TaskProfileDto.java

@@ -38,6 +38,8 @@
 import com.google.gson.Gson;
 import lombok.Data;
 
+import java.util.stream.Collectors;
+
 import static java.util.Objects.requireNonNull;
 import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_ADDR;
 import static org.apache.inlong.agent.constant.TaskConstants.SYNC_SEND_OPEN;
@@ -50,6 +52,7 @@ public class TaskProfileDto {
  public static final String DEFAULT_KAFKA_TASK = "org.apache.inlong.agent.plugin.task.KafkaTask";
  public static final String DEFAULT_PULSAR_TASK = "org.apache.inlong.agent.plugin.task.PulsarTask";
  public static final String DEFAULT_MONGODB_TASK = "org.apache.inlong.agent.plugin.task.MongoDBTask";
+ public static final String DEFAULT_POSTGRESQL_TASK = "org.apache.inlong.agent.plugin.task.PostgreSQLTask";
  public static final String DEFAULT_CHANNEL = "org.apache.inlong.agent.plugin.channel.MemoryChannel";
  public static final String MANAGER_JOB = "MANAGER_JOB";
  public static final String DEFAULT_DATA_PROXY_SINK = "org.apache.inlong.agent.plugin.sinks.ProxySink";
@@ -242,11 +245,14 @@ private static PostgreSQLTask getPostgresTask(DataConfig dataConfigs) {
  postgreSQLTask.setHostname(config.getHostname());
  postgreSQLTask.setPort(config.getPort());
  postgreSQLTask.setDbname(config.getDatabase());
- postgreSQLTask.setServername(config.getSchema());
- postgreSQLTask.setPluginname(config.getDecodingPluginName());
- postgreSQLTask.setTableNameList(config.getTableNameList());
+ postgreSQLTask.setSchemaIncludeList(config.getSchema());
+ postgreSQLTask.setPluginName(config.getDecodingPluginName());
+ // Each identifier is of the form schemaName.tableName and connected with ","
+ postgreSQLTask.setTableIncludeList(
+ config.getTableNameList().stream().map(tableName -> config.getSchema() + "." + tableName).collect(
+ Collectors.joining(",")));
  postgreSQLTask.setServerTimeZone(config.getServerTimeZone());
- postgreSQLTask.setScanStartupMode(config.getScanStartupMode());
+ postgreSQLTask.setSnapshotMode(config.getScanStartupMode());
  postgreSQLTask.setPrimaryKey(config.getPrimaryKey());
 
  return postgreSQLTask;
@@ -475,6 +481,7 @@ public static TaskProfile convertToTaskProfile(DataConfig dataConfig) {
  profileDto.setTask(task);
  break;
  case POSTGRES:
+ task.setTaskClass(DEFAULT_POSTGRESQL_TASK);
  PostgreSQLTask postgreSQLTask = getPostgresTask(dataConfig);
  task.setPostgreSQLTask(postgreSQLTask);
  task.setSource(POSTGRESQL_SOURCE);

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/instance/PostgreSQLInstance.java

@@ -0,0 +1,29 @@
+/*
+ * 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.instance;
+
+import org.apache.inlong.agent.conf.InstanceProfile;
+import org.apache.inlong.agent.constant.TaskConstants;
+
+public class PostgreSQLInstance extends CommonInstance {
+
+ @Override
+ public void setInodeInfo(InstanceProfile profile) {
+ profile.set(TaskConstants.INODE_INFO, "");
+ }
+}

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

@@ -17,67 +17,173 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+import org.apache.inlong.agent.common.AgentThreadFactory;
+import org.apache.inlong.agent.conf.AgentConfiguration;
 import org.apache.inlong.agent.conf.InstanceProfile;
 import org.apache.inlong.agent.conf.TaskProfile;
-import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.constant.AgentConstants;
+import org.apache.inlong.agent.constant.TaskConstants;
+import org.apache.inlong.agent.except.FileException;
 import org.apache.inlong.agent.plugin.file.Reader;
 import org.apache.inlong.agent.plugin.sources.file.AbstractSource;
-import org.apache.inlong.agent.plugin.sources.reader.PostgreSQLReader;
 
+import io.debezium.connector.postgresql.PostgresConnector;
+import io.debezium.connector.postgresql.PostgresConnectorConfig;
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.DebeziumEngine.RecordCommitter;
+import io.debezium.engine.format.Json;
+import io.debezium.engine.spi.OffsetCommitPolicy;
+import org.apache.kafka.connect.storage.FileOffsetBackingStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_DBNAME;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_HOSTNAME;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_PASSWORD;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_PLUGIN_NAME;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_PORT;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_SCHEMA_INCLUDE_LIST;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_TABLE_INCLUDE_LIST;
+import static org.apache.inlong.agent.constant.TaskConstants.TASK_POSTGRES_USER;
 
-/**
- * PostgreSQL source, split PostgreSQL source job into multi readers
- */
 public class PostgreSQLSource extends AbstractSource {
 
  private static final Logger LOGGER = LoggerFactory.getLogger(PostgreSQLSource.class);
+ private static final Integer DEBEZIUM_QUEUE_SIZE = 100;
+ private ExecutorService executor;
+ public InstanceProfile profile;
+ private BlockingQueue<SourceData> debeziumQueue;
+ private final Properties props = new Properties();
+ private String snapshotMode;
+ private String pluginName;
+ private String dbName;
+ private String tableName;
+
+ private boolean isRestoreFromDB = false;
 
  public PostgreSQLSource() {
 
  }
 
  @Override
- public List<Reader> split(TaskProfile conf) {
- PostgreSQLReader postgreSQLReader = new PostgreSQLReader();
- List<Reader> readerList = new ArrayList<>();
- readerList.add(postgreSQLReader);
- sourceMetric.sourceSuccessCount.incrementAndGet();
- return readerList;
+ protected void initSource(InstanceProfile profile) {
+ try {
+ LOGGER.info("PostgreSQLSource init: {}", profile.toJsonStr());
+ debeziumQueue = new LinkedBlockingQueue<>(DEBEZIUM_QUEUE_SIZE);
+ pluginName = profile.get(TASK_POSTGRES_PLUGIN_NAME);
+ dbName = profile.get(TASK_POSTGRES_DBNAME);
+ tableName = profile.get(TASK_POSTGRES_TABLE_INCLUDE_LIST);
+ snapshotMode = profile.get(TaskConstants.TASK_POSTGRES_SNAPSHOT_MODE, "initial");
+
+ props.setProperty("name", "PostgreSQL-" + instanceId);
+ props.setProperty("connector.class", PostgresConnector.class.getName());
+ props.setProperty("offset.storage", FileOffsetBackingStore.class.getName());
+ String agentPath = AgentConfiguration.getAgentConf()
+ .get(AgentConstants.AGENT_HOME, AgentConstants.DEFAULT_AGENT_HOME);
+ String offsetPath = agentPath + "/" + getThreadName() + "offset.dat";
+ props.setProperty("offset.storage.file.filename", offsetPath);
+
+ props.setProperty(String.valueOf(PostgresConnectorConfig.HOSTNAME), profile.get(TASK_POSTGRES_HOSTNAME));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.PORT), profile.get(TASK_POSTGRES_PORT));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.USER), profile.get(TASK_POSTGRES_USER));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.PASSWORD), profile.get(TASK_POSTGRES_PASSWORD));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.DATABASE_NAME), profile.get(TASK_POSTGRES_DBNAME));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.SERVER_NAME), getThreadName());
+ props.setProperty(String.valueOf(PostgresConnectorConfig.SCHEMA_INCLUDE_LIST),
+ profile.get(TASK_POSTGRES_SCHEMA_INCLUDE_LIST));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.TABLE_INCLUDE_LIST),
+ profile.get(TASK_POSTGRES_TABLE_INCLUDE_LIST));
+ props.setProperty(String.valueOf(PostgresConnectorConfig.PLUGIN_NAME), pluginName);
+ props.setProperty(String.valueOf(PostgresConnectorConfig.SNAPSHOT_MODE), snapshotMode);
+
+ executor = Executors.newSingleThreadExecutor();
+ executor.execute(startDebeziumEngine());
+
+ } catch (Exception ex) {
+ stopRunning();
+ throw new FileException("error init stream for " + instanceId, ex);
+ }
+ }
+
+ private Runnable startDebeziumEngine() {
+ return () -> {
+ AgentThreadFactory.nameThread(getThreadName() + "debezium");
+ try (DebeziumEngine<ChangeEvent<String, String>> debeziumEngine = DebeziumEngine.create(Json.class)
+ .using(props)
+ .using(OffsetCommitPolicy.always())
+ .notifying(this::handleConsumerEvent)
+ .build()) {
+
+ debeziumEngine.run();
+ } catch (Throwable e) {
+ LOGGER.error("do run error in postgres debezium: ", e);
+ }
+ };
+ }
+
+ private void handleConsumerEvent(List<ChangeEvent<String, String>> records,
+ RecordCommitter<ChangeEvent<String, String>> committer) throws InterruptedException {
+ boolean offerSuc = false;
+ for (ChangeEvent<String, String> record : records) {
+ SourceData sourceData = new SourceData(record.value().getBytes(StandardCharsets.UTF_8), "0");
+ while (isRunnable() && !offerSuc) {
+ offerSuc = debeziumQueue.offer(sourceData, 1, TimeUnit.SECONDS);
+ }
+ committer.markProcessed(record);
+ }
+ committer.markBatchFinished();
  }
 
  @Override
- protected String getThreadName() {
+ public List<Reader> split(TaskProfile conf) {
  return null;
  }
 
  @Override
- protected void initSource(InstanceProfile profile) {
-
+ protected String getThreadName() {
+ return "postgres-source-" + taskId + "-" + instanceId;
  }
 
  @Override
  protected void printCurrentState() {
-
+ LOGGER.info("postgres databases is {} and table is {}", dbName, tableName);
  }
 
  @Override
  protected boolean doPrepareToRead() {
- return false;
+ return true;
  }
 
  @Override
  protected List<SourceData> readFromSource() {
- return null;
- }
-
- @Override
- public Message read() {
- return null;
+ List<SourceData> dataList = new ArrayList<>();
+ try {
+ int size = 0;
+ while (size < BATCH_READ_LINE_TOTAL_LEN) {
+ SourceData sourceData = debeziumQueue.poll(1, TimeUnit.SECONDS);
+ if (sourceData != null) {
+ LOGGER.info("readFromSource: {}", sourceData.getData());
+ size += sourceData.getData().length;
+ dataList.add(sourceData);
+ } else {
+ break;
+ }
+ }
+ } catch (InterruptedException e) {
+ LOGGER.error("poll {} data from debezium queue interrupted.", instanceId);
+ }
+ return dataList;
  }
 
  @Override
@@ -87,16 +193,12 @@ protected boolean isRunnable() {
 
  @Override
  protected void releaseSource() {
-
- }
-
- @Override
- public boolean sourceFinish() {
- return false;
+ LOGGER.info("release postgres source");
+ executor.shutdownNow();
  }
 
  @Override
  public boolean sourceExist() {
- return false;
+ return true;
  }
 }