[INLONG-9310][Agent] Add extended handler in file source

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/filecollect/ProxyMessage.java

@@ -27,7 +27,7 @@
 import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_STREAM_ID;
 
 /**
- * Bus message with body, header, inlongGroupId and inlongStreamId.
+ * proxy message with body, header, inlongGroupId and inlongStreamId.
  */
 public class ProxyMessage implements Message {
 

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/filecollect/ProxyMessageCache.java

@@ -59,7 +59,6 @@ public class ProxyMessageCache {
     private final int cacheTimeout;
     // streamId -> list of proxyMessage
     private final ConcurrentHashMap<String, LinkedBlockingQueue<ProxyMessage>> messageQueueMap;
-    // private final LinkedBlockingQueue<ProxyMessage> messageQueue;
     private final AtomicLong cacheSize = new AtomicLong(0);
     private long lastPrintTime = 0;
     private long dataTime;
@@ -77,7 +76,6 @@ public ProxyMessageCache(InstanceProfile instanceProfile, String groupId, String
                 DEFAULT_PROXY_INLONG_STREAM_ID_QUEUE_MAX_NUMBER);
         this.cacheTimeout = instanceProfile.getInt(PROXY_PACKAGE_MAX_TIMEOUT_MS, DEFAULT_PROXY_PACKAGE_MAX_TIMEOUT_MS);
         messageQueueMap = new ConcurrentHashMap<>();
-        // this.messageQueue = new LinkedBlockingQueue<>(maxQueueNumber);
         try {
             dataTime = DateTransUtils.timeStrConvertToMillSec(instanceProfile.getSourceDataTime(),
                     instanceProfile.get(TASK_CYCLE_UNIT));

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

@@ -31,6 +31,7 @@
 import org.apache.inlong.agent.plugin.Message;
 import org.apache.inlong.agent.plugin.file.Reader;
 import org.apache.inlong.agent.plugin.sources.file.AbstractSource;
+import org.apache.inlong.agent.plugin.sources.file.extend.ExtendedHandler;
 import org.apache.inlong.agent.plugin.sources.reader.file.KubernetesMetadataProvider;
 import org.apache.inlong.agent.plugin.utils.file.FileDataUtils;
 import org.apache.inlong.agent.utils.AgentUtils;
@@ -134,6 +135,7 @@ private class SourceData {
     private volatile boolean running = false;
     private long dataTime = 0;
     private volatile long emptyCount = 0;
+    private ExtendedHandler extendedHandler;
 
     public LogFileSource() {
         OffsetManager.init();
@@ -159,6 +161,7 @@ public void init(InstanceProfile profile) {
             queue = new LinkedBlockingQueue<>(CACHE_QUEUE_SIZE);
             dataTime = DateTransUtils.timeStrConvertToMillSec(profile.getSourceDataTime(),
                     profile.get(TASK_CYCLE_UNIT));
+            extendedHandler = new ExtendedHandler(profile);
             try {
                 registerMeta(profile);
             } catch (Exception ex) {
@@ -354,6 +357,7 @@ private Message createMessage(SourceData sourceData) {
         header.put(PROXY_KEY_DATA, proxyPartitionKey);
         header.put(OFFSET, sourceData.offset.toString());
         header.put(PROXY_KEY_STREAM_ID, inlongStreamId);
+        extendedHandler.dealWithHeader(header, sourceData.getData().getBytes(StandardCharsets.UTF_8));
         AuditUtils.add(AuditUtils.AUDIT_ID_AGENT_READ_SUCCESS, inlongGroupId, header.get(PROXY_KEY_STREAM_ID),
                 dataTime, 1, msgWithMetaData.length());
         Message finalMsg = new DefaultMessage(msgWithMetaData.getBytes(StandardCharsets.UTF_8), header);

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/file/extend/ExtendedHandler.java

@@ -0,0 +1,37 @@
+/*
+ * 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.file.extend;
+
+import org.apache.inlong.agent.conf.InstanceProfile;
+
+import java.util.Map;
+
+public class ExtendedHandler {
+
+    public ExtendedHandler(InstanceProfile profile) {
+
+    }
+
+    public void dealWithHeader(Map<String, String> header, byte[] body) {
+
+    }
+
+    public static class Constants {
+
+    }
+}