[INLONG-9134][Agent] Add file related utils

justinwwhuang · Oct 27, 2023 · 517ad53 · 517ad53
1 parent 2f90ca4
commit 517ad53
Show file tree

Hide file tree

Showing 14 changed files with 1,866 additions and 23 deletions.
diff --git a/...t-common/src/main/java/org/apache/inlong/agent/message/filecollect/ProxyMessageCache.java b/...t-common/src/main/java/org/apache/inlong/agent/message/filecollect/ProxyMessageCache.java
@@ -42,7 +42,7 @@
 import static org.apache.inlong.agent.constant.CommonConstants.PROXY_PACKAGE_MAX_TIMEOUT_MS;
 
 /**
- * Handle List of BusMessage, which belong to the same stream id.
+ * Handle List of Proxy Message, which belong to the same stream id.
  */
 public class ProxyMessageCache {
 
@@ -67,17 +67,13 @@ public class ProxyMessageCache {
      */
     private Map<String, String> extraMap = new HashMap<>();
 
-    /**
-     * Init PackBusMessage
-     */
     public ProxyMessageCache(InstanceProfile instanceProfile, String groupId, String streamId) {
         this.taskId = instanceProfile.getTaskId();
         this.instanceId = instanceProfile.getInstanceId();
         this.maxPackSize = instanceProfile.getInt(PROXY_PACKAGE_MAX_SIZE, DEFAULT_PROXY_PACKAGE_MAX_SIZE);
         this.maxQueueNumber = instanceProfile.getInt(PROXY_INLONG_STREAM_ID_QUEUE_MAX_NUMBER,
                 DEFAULT_PROXY_INLONG_STREAM_ID_QUEUE_MAX_NUMBER);
         this.cacheTimeout = instanceProfile.getInt(PROXY_PACKAGE_MAX_TIMEOUT_MS, DEFAULT_PROXY_PACKAGE_MAX_TIMEOUT_MS);
-        // double size of package
         this.messageQueue = new LinkedBlockingQueue<>(maxQueueNumber);
         this.groupId = groupId;
         this.streamId = streamId;

diff --git a/inlong-agent/agent-plugins/pom.xml b/inlong-agent/agent-plugins/pom.xml
@@ -32,6 +32,7 @@
     <properties>
         <inlong.root.dir>${project.parent.parent.basedir}</inlong.root.dir>
         <debezium.version>1.8.0.Final</debezium.version>
+        <darwinsys.version>1.5.1</darwinsys.version>
     </properties>
 
     <dependencies>
@@ -202,6 +203,11 @@
             <artifactId>powermock-api-mockito2</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.darwinsys</groupId>
+            <artifactId>hirondelle-date4j</artifactId>
+            <version>${darwinsys.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.fabric8</groupId>
             <artifactId>kubernetes-client</artifactId>

diff --git a/...nt/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/TextFileSource.java b/...nt/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/TextFileSource.java
@@ -21,8 +21,8 @@
 import org.apache.inlong.agent.plugin.Reader;
 import org.apache.inlong.agent.plugin.sources.reader.file.FileReaderOperator;
 import org.apache.inlong.agent.plugin.sources.reader.file.TriggerFileReader;
-import org.apache.inlong.agent.plugin.utils.FileDataUtils;
 import org.apache.inlong.agent.plugin.utils.PluginUtils;
+import org.apache.inlong.agent.plugin.utils.file.FileDataUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

diff --git a/.../src/main/java/org/apache/inlong/agent/plugin/sources/reader/file/FileReaderOperator.java b/.../src/main/java/org/apache/inlong/agent/plugin/sources/reader/file/FileReaderOperator.java
@@ -29,7 +29,7 @@
 import org.apache.inlong.agent.metrics.audit.AuditUtils;
 import org.apache.inlong.agent.plugin.Message;
 import org.apache.inlong.agent.plugin.sources.reader.AbstractReader;
-import org.apache.inlong.agent.plugin.utils.FileDataUtils;
+import org.apache.inlong.agent.plugin.utils.file.FileDataUtils;
 import org.apache.inlong.agent.utils.AgentUtils;
 
 import com.google.gson.Gson;

diff --git a/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/MetaDataUtils.java b/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/MetaDataUtils.java
@@ -17,7 +17,7 @@
 
 package org.apache.inlong.agent.plugin.utils;
 
-import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.conf.AbstractConfiguration;
 import org.apache.inlong.agent.constant.CommonConstants;
 
 import com.google.gson.Gson;
@@ -33,12 +33,12 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-import static org.apache.inlong.agent.constant.JobConstants.JOB_FILE_META_FILTER_BY_LABELS;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_FILE_PROPERTIES;
 import static org.apache.inlong.agent.constant.KubernetesConstants.CONTAINER_ID;
 import static org.apache.inlong.agent.constant.KubernetesConstants.CONTAINER_NAME;
 import static org.apache.inlong.agent.constant.KubernetesConstants.NAMESPACE;
 import static org.apache.inlong.agent.constant.KubernetesConstants.POD_NAME;
+import static org.apache.inlong.agent.constant.TaskConstants.JOB_FILE_META_FILTER_BY_LABELS;
+import static org.apache.inlong.agent.constant.TaskConstants.JOB_FILE_PROPERTIES;
 
 /**
  * Metadata utils
@@ -89,21 +89,21 @@ public static Map<String, String> getLogInfo(String fileName) {
      *
      * get labels of pod
      */
-    public static Map<String, String> getPodLabels(JobProfile jobProfile) {
-        if (Objects.isNull(jobProfile) || !jobProfile.hasKey(JOB_FILE_META_FILTER_BY_LABELS)) {
+    public static Map<String, String> getPodLabels(AbstractConfiguration taskProfile) {
+        if (Objects.isNull(taskProfile) || !taskProfile.hasKey(JOB_FILE_META_FILTER_BY_LABELS)) {
             return new HashMap<>();
         }
-        String labels = jobProfile.get(JOB_FILE_META_FILTER_BY_LABELS);
+        String labels = taskProfile.get(JOB_FILE_META_FILTER_BY_LABELS);
         Type type = new TypeToken<HashMap<String, String>>() {
         }.getType();
         return GSON.fromJson(labels, type);
     }
 
-    public static List<String> getNamespace(JobProfile jobProfile) {
-        if (Objects.isNull(jobProfile) || !jobProfile.hasKey(JOB_FILE_PROPERTIES)) {
+    public static List<String> getNamespace(AbstractConfiguration taskProfile) {
+        if (Objects.isNull(taskProfile) || !taskProfile.hasKey(JOB_FILE_PROPERTIES)) {
             return null;
         }
-        String property = jobProfile.get(JOB_FILE_PROPERTIES);
+        String property = taskProfile.get(JOB_FILE_PROPERTIES);
         Type type = new TypeToken<HashMap<Integer, String>>() {
         }.getType();
         Map<String, String> properties = GSON.fromJson(property, type);
@@ -120,11 +120,11 @@ public static List<String> getNamespace(JobProfile jobProfile) {
      *
      * get name of pod
      */
-    public static String getPodName(JobProfile jobProfile) {
-        if (Objects.isNull(jobProfile) || !jobProfile.hasKey(JOB_FILE_PROPERTIES)) {
+    public static String getPodName(AbstractConfiguration taskProfile) {
+        if (Objects.isNull(taskProfile) || !taskProfile.hasKey(JOB_FILE_PROPERTIES)) {
             return null;
         }
-        String property = jobProfile.get(JOB_FILE_PROPERTIES);
+        String property = taskProfile.get(JOB_FILE_PROPERTIES);
         Type type = new TypeToken<HashMap<Integer, String>>() {
         }.getType();
         Map<String, String> properties = GSON.fromJson(property, type);