[INLONG-1641][Agent] introduce a Message filter #1641 (#1644)

* [INLONG-1592][1641] Agent introduce a Message filter #1641

* [INLONG-1641][Agent] Agent introduce a Message filter

* [INLONG-1641][Agent] Agent introduce a Message filter

* [INLONG-1641][Agent] Agent introduce a Message filter

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/MessageFilter.java

@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+public interface MessageFilter {
+
+    /**
+     * split a message to get tid string
+     * used when the file is separated with different tid
+     * @param message the input message
+     * @param fieldSplitter fieldSplitter used when split a line
+     * @return
+     */
+    String filterTid(Message message, byte[] fieldSplitter);
+}

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/filter/DefaultMessageFilter.java

@@ -0,0 +1,38 @@
+/*
+ * 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.filter;
+
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.MessageFilter;
+import org.apache.inlong.agent.utils.ByteUtil;
+
+/**
+ * filter message to get tid
+ * use the first word to set tid string
+ */
+public class DefaultMessageFilter implements MessageFilter {
+
+    public static final int TID_INDEX = 0;
+    public static final int FIELDS_LIMIT = 2;
+
+    @Override
+    public String filterTid(Message message, byte[] fieldSplitter) {
+        byte[] body = message.getBody();
+        return new String(ByteUtil.split(body, fieldSplitter, FIELDS_LIMIT)[TID_INDEX]);
+    }
+}

inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/filter/TestTidFilter.java

@@ -0,0 +1,56 @@
+/*
+ * 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.filter;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.HashMap;
+import org.apache.inlong.agent.message.ProxyMessage;
+import org.apache.inlong.agent.plugin.AgentBaseTestsHelper;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestTidFilter {
+
+    private static AgentBaseTestsHelper helper;
+    private static Path testPath;
+
+    @BeforeClass
+    public static void setup() {
+        helper = new AgentBaseTestsHelper(TestDateFormatRegex.class.getName()).setupAgentHome();
+        testPath = helper.getTestRootDir();
+    }
+
+    @AfterClass
+    public static void teardown() {
+        helper.teardownAgentHome();
+    }
+
+    @Test
+    public void testFilterTid() {
+        DefaultMessageFilter messageFilter = new DefaultMessageFilter();
+        ProxyMessage proxyMessage = new ProxyMessage("tid|this is a line of file".getBytes(
+            StandardCharsets.UTF_8), new HashMap<>());
+        String s = messageFilter.filterTid(proxyMessage, "|".getBytes(StandardCharsets.UTF_8));
+        Assert.assertEquals(s, "tid");
+    }
+
+}
+