[INLONG-11571][Agent] Add classes for actual collection of COS source (…

…#11572) * [INLONG-11571][Agent] Add classes for actual collection of COS source * [INLONG-11571][Agent] Delete invalid comments * Update inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/cos/COSTask.java Co-authored-by: AloysZhang <[email protected]> * [INLONG-11571][Agent] Modify code based on comments --------- Co-authored-by: AloysZhang <[email protected]>
apache · Dec 4, 2024 · 7ccbd97 · 7ccbd97
1 parent 610594d
commit 7ccbd97
Show file tree

Hide file tree

Showing 16 changed files with 1,025 additions and 17 deletions.
diff --git a/inlong-agent/agent-plugins/pom.xml b/inlong-agent/agent-plugins/pom.xml
@@ -92,6 +92,11 @@
             <artifactId>kafka-clients</artifactId>
             <version>${kafka.clients.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.qcloud</groupId>
+            <artifactId>cos_api</artifactId>
+            <version>${cos.sdk.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.debezium</groupId>
             <artifactId>debezium-connector-mysql</artifactId>

diff --git a/...nt/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java b/...nt/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
@@ -28,6 +28,7 @@
 import org.apache.inlong.agent.utils.HttpManager;
 import org.apache.inlong.agent.utils.ThreadUtils;
 import org.apache.inlong.common.enums.PullJobTypeEnum;
+import org.apache.inlong.common.enums.TaskTypeEnum;
 import org.apache.inlong.common.pojo.agent.AgentConfigInfo;
 import org.apache.inlong.common.pojo.agent.AgentConfigRequest;
 import org.apache.inlong.common.pojo.agent.AgentResponseCode;
@@ -237,7 +238,7 @@ private DataConfig getTestDataConfig(int taskId, String pattern, boolean retry,
         dataConfig.setInlongGroupId("devcloud_group_id");
         dataConfig.setInlongStreamId("devcloud_stream_id");
         dataConfig.setDataReportType(0);
-        dataConfig.setTaskType(3);
+        dataConfig.setTaskType(TaskTypeEnum.FILE.getType());
         dataConfig.setTaskId(taskId);
         dataConfig.setState(state);
         dataConfig.setTimeZone("GMT+8:00");

diff --git a/...gent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/instance/COSInstance.java b/...gent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/instance/COSInstance.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.plugin.instance;
+
+import org.apache.inlong.agent.conf.InstanceProfile;
+import org.apache.inlong.agent.constant.TaskConstants;
+
+import java.io.IOException;
+
+/**
+ * cos instance contains source and sink.
+ * main job is to read from source and write to sink
+ */
+public class COSInstance extends CommonInstance {
+
+    @Override
+    public void setInodeInfo(InstanceProfile profile) throws IOException {
+        profile.set(TaskConstants.INODE_INFO, "");
+    }
+}
diff --git a/...t/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/instance/CommonInstance.java b/...t/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/instance/CommonInstance.java
@@ -122,7 +122,7 @@ public void notifyDestroy() {
 
     @Override
     public void run() {
-        Thread.currentThread().setName("file-instance-core-" + getTaskId() + "-" + getInstanceId());
+        Thread.currentThread().setName("instance-core-" + getTaskId() + "-" + getInstanceId());
         running = true;
         try {
             doRun();

diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
@@ -196,18 +196,18 @@ public void destroy() {
         }
         Long start = AgentUtils.getCurrentTime();
         shutdown = true;
+        senderManager.Stop();
+        LOGGER.info("destroy proxySink, wait for sender close {} ms instance {}", AgentUtils.getCurrentTime() - start,
+                profile.getInstanceId());
+        start = AgentUtils.getCurrentTime();
         while (running || offsetRunning) {
             AgentUtils.silenceSleepInMs(LOOP_WAIT_TIME_MS);
         }
-        LOGGER.info("destroy proxySink wait run elapse {} ms instance {}", AgentUtils.getCurrentTime() - start,
-                profile.getInstanceId());
-        start = AgentUtils.getCurrentTime();
-        senderManager.Stop();
-        LOGGER.info("destroy proxySink wait sender elapse {} ms instance {}", AgentUtils.getCurrentTime() - start,
+        LOGGER.info("destroy proxySink, wait for run close {} ms instance {}", AgentUtils.getCurrentTime() - start,
                 profile.getInstanceId());
         start = AgentUtils.getCurrentTime();
         clearOffset();
-        LOGGER.info("destroy proxySink wait offset elapse {} ms instance {}", AgentUtils.getCurrentTime() - start,
+        LOGGER.info("destroy proxySink, wait for offset clear {} ms instance {}", AgentUtils.getCurrentTime() - start,
                 profile.getInstanceId());
         LOGGER.info("destroy sink {} end", sourceName);
     }