[INLONG-10010][Agent] Adjust source encapsulation, keep public initia…

…lization in the base class init, and place specific source related initialization in the subclass's initSource (#10011)
apache · Apr 19, 2024 · a3cc501 · a3cc501
1 parent b206b83
commit a3cc501
Show file tree

Hide file tree

Showing 11 changed files with 50 additions and 10 deletions.
diff --git a/...agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/DatabaseSqlSource.java b/...agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/DatabaseSqlSource.java
@@ -17,6 +17,7 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+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.plugin.file.Reader;
@@ -93,6 +94,11 @@ protected String getThreadName() {
         return null;
     }
 
+    @Override
+    protected void initSource(InstanceProfile profile) {
+
+    }
+
     @Override
     protected void printCurrentState() {
 

diff --git a/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java b/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java
@@ -69,11 +69,9 @@ public KafkaSource() {
     }
 
     @Override
-    public void init(InstanceProfile profile) {
+    protected void initSource(InstanceProfile profile) {
         try {
             LOGGER.info("KafkaSource init: {}", profile.toJsonStr());
-            this.profile = profile;
-            super.init(profile);
             topic = profile.getInstanceId();
             props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, profile.get(TASK_KAFKA_BOOTSTRAP_SERVERS));
             props.put(ConsumerConfig.GROUP_ID_CONFIG, taskId);

diff --git a/...ent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/LogFileSource.java b/...ent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/LogFileSource.java
@@ -65,10 +65,9 @@ public LogFileSource() {
     }
 
     @Override
-    public void init(InstanceProfile profile) {
+    protected void initSource(InstanceProfile profile) {
         try {
             LOGGER.info("LogFileSource init: {}", profile.toJsonStr());
-            super.init(profile);
             fileName = profile.getInstanceId();
             bufferToReadFile = new byte[SIZE_OF_BUFFER_TO_READ_FILE];
             isIncrement = isIncrement(profile);

diff --git a/...ent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MongoDBSource.java b/...ent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MongoDBSource.java
@@ -66,11 +66,9 @@ public MongoDBSource() {
     }
 
     @Override
-    public void init(InstanceProfile profile) {
+    protected void initSource(InstanceProfile profile) {
         try {
             LOGGER.info("MongoDBSource init: {}", profile.toJsonStr());
-            this.profile = profile;
-            super.init(profile);
             debeziumQueue = new LinkedBlockingQueue<>(DEBEZIUM_QUEUE_SIZE);
             database = profile.get(TaskConstants.TASK_MONGO_DATABASE_INCLUDE_LIST);
             collection = profile.get(TaskConstants.TASK_MONGO_COLLECTION_INCLUDE_LIST);

diff --git a/...-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java b/...-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java
@@ -17,6 +17,7 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+import org.apache.inlong.agent.conf.InstanceProfile;
 import org.apache.inlong.agent.conf.TaskProfile;
 import org.apache.inlong.agent.constant.CommonConstants;
 import org.apache.inlong.agent.plugin.Message;
@@ -84,6 +85,11 @@ protected String getThreadName() {
         return null;
     }
 
+    @Override
+    protected void initSource(InstanceProfile profile) {
+
+    }
+
     @Override
     protected void printCurrentState() {
 

diff --git a/...gent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/OracleSource.java b/...gent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/OracleSource.java
@@ -17,6 +17,7 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+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.plugin.file.Reader;
@@ -53,6 +54,11 @@ protected String getThreadName() {
         return null;
     }
 
+    @Override
+    protected void initSource(InstanceProfile profile) {
+
+    }
+
     @Override
     protected void printCurrentState() {
 

diff --git a/.../agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/PostgreSQLSource.java b/.../agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/PostgreSQLSource.java
@@ -17,6 +17,7 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+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.plugin.file.Reader;
@@ -54,6 +55,11 @@ protected String getThreadName() {
         return null;
     }
 
+    @Override
+    protected void initSource(InstanceProfile profile) {
+
+    }
+
     @Override
     protected void printCurrentState() {
 

diff --git a/...gent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/PulsarSource.java b/...gent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/PulsarSource.java
@@ -63,10 +63,9 @@ public PulsarSource() {
     }
 
     @Override
-    public void init(InstanceProfile profile) {
+    protected void initSource(InstanceProfile profile) {
         try {
             LOGGER.info("PulsarSource init: {}", profile.toJsonStr());
-            super.init(profile);
             topic = profile.getInstanceId();
             serviceUrl = profile.get(TASK_PULSAR_SERVICE_URL);
             subscription = profile.get(TASK_PULSAR_SUBSCRIPTION, PULSAR_SUBSCRIPTION_PREFIX + inlongStreamId);

diff --git a/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/RedisSource.java b/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/RedisSource.java
@@ -17,6 +17,7 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+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.plugin.file.Reader;
@@ -54,6 +55,11 @@ protected String getThreadName() {
         return null;
     }
 
+    @Override
+    protected void initSource(InstanceProfile profile) {
+
+    }
+
     @Override
     protected void printCurrentState() {
 

diff --git a/...t/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/SQLServerSource.java b/...t/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/SQLServerSource.java
@@ -17,6 +17,7 @@
 
 package org.apache.inlong.agent.plugin.sources;
 
+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.plugin.file.Reader;
@@ -53,6 +54,11 @@ protected String getThreadName() {
         return null;
     }
 
+    @Override
+    protected void initSource(InstanceProfile profile) {
+
+    }
+
     @Override
     protected void printCurrentState() {
 

diff --git a/...ent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/file/AbstractSource.java b/...ent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/file/AbstractSource.java
@@ -20,6 +20,7 @@
 import org.apache.inlong.agent.common.AgentThreadFactory;
 import org.apache.inlong.agent.conf.InstanceProfile;
 import org.apache.inlong.agent.conf.OffsetProfile;
+import org.apache.inlong.agent.conf.TaskProfile;
 import org.apache.inlong.agent.constant.CycleUnitType;
 import org.apache.inlong.agent.constant.TaskConstants;
 import org.apache.inlong.agent.core.task.MemoryManager;
@@ -29,6 +30,7 @@
 import org.apache.inlong.agent.metrics.AgentMetricItemSet;
 import org.apache.inlong.agent.metrics.audit.AuditUtils;
 import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.file.Reader;
 import org.apache.inlong.agent.plugin.file.Source;
 import org.apache.inlong.agent.plugin.sources.file.extend.ExtendedHandler;
 import org.apache.inlong.agent.utils.AgentUtils;
@@ -128,8 +130,11 @@ public void init(InstanceProfile profile) {
         initOffset();
         registerMetric();
         initExtendHandler();
+        initSource(profile);
     }
 
+    protected abstract void initSource(InstanceProfile profile);
+
     protected void initOffset() {
         offsetProfile = OffsetManager.getInstance().getOffset(taskId, instanceId);
     }
@@ -396,4 +401,9 @@ public boolean sourceFinish() {
         }
         return emptyCount > EMPTY_CHECK_COUNT_AT_LEAST;
     }
+
+    @Override
+    public List<Reader> split(TaskProfile conf) {
+        return null;
+    }
 }