You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2022/05/13 09:36:41 UTC
[incubator-inlong] branch master updated: [INLONG-4193][Agent] Add Java docs (#4196)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new a208d5d1b [INLONG-4193][Agent] Add Java docs (#4196)
a208d5d1b is described below
commit a208d5d1beb794037e2174d8886c6e4656d0d712
Author: xueyingzhang <86...@users.noreply.github.com>
AuthorDate: Fri May 13 17:36:35 2022 +0800
[INLONG-4193][Agent] Add Java docs (#4196)
---
.../apache/inlong/agent/cache/LocalFileCache.java | 12 +--
.../apache/inlong/agent/checkpoint/Checkpoint.java | 31 -------
.../apache/inlong/agent/common/AbstractDaemon.java | 19 +++--
.../inlong/agent/common/AgentThreadFactory.java | 8 +-
.../org/apache/inlong/agent/common/Service.java | 3 +
.../inlong/agent/conf/AbstractConfiguration.java | 66 ++++++++-------
.../inlong/agent/conf/AgentConfiguration.java | 17 ++--
.../org/apache/inlong/agent/conf/JobProfile.java | 9 +-
.../apache/inlong/agent/conf/ProfileFetcher.java | 11 ++-
.../inlong/agent/constant/AgentConstants.java | 3 +
.../inlong/agent/constant/CommonConstants.java | 3 +
.../main/java/org/apache/inlong/agent/db/Db.java | 24 ++----
.../org/apache/inlong/agent/db/JobProfileDb.java | 50 +++++++----
.../org/apache/inlong/agent/db/KeyValueEntity.java | 3 +-
.../org/apache/inlong/agent/db/LocalProfile.java | 13 ++-
.../org/apache/inlong/agent/db/RocksDbImp.java | 22 +++--
.../apache/inlong/agent/db/TriggerProfileDb.java | 14 ++--
.../inlong/agent/message/DefaultMessage.java | 7 +-
.../apache/inlong/agent/message/ProxyMessage.java | 21 +++--
.../inlong/agent/metrics/MetricException.java | 3 +
.../inlong/agent/metrics/MetricSnapshot.java | 3 +
.../org/apache/inlong/agent/metrics/Metrics.java | 6 --
.../inlong/agent/metrics/MetricsRegister.java | 22 +++--
.../inlong/agent/metrics/audit/AuditUtils.java | 2 -
.../inlong/agent/metrics/counter/Counter.java | 3 +
.../inlong/agent/metrics/counter/CounterInt.java | 3 +
.../apache/inlong/agent/metrics/gauge/Gauge.java | 6 ++
.../inlong/agent/metrics/gauge/GaugeInt.java | 3 +
.../inlong/agent/metrics/gauge/GaugeLong.java | 3 +
.../inlong/agent/metrics/meta/MetricMeta.java | 22 +++--
.../inlong/agent/metrics/meta/MetricsMeta.java | 10 ++-
.../org/apache/inlong/agent/plugin/Channel.java | 11 ---
.../apache/inlong/agent/plugin/MessageFilter.java | 6 +-
.../org/apache/inlong/agent/plugin/Reader.java | 6 +-
.../java/org/apache/inlong/agent/plugin/Sink.java | 3 +-
.../org/apache/inlong/agent/plugin/Source.java | 5 +-
.../java/org/apache/inlong/agent/plugin/Stage.java | 2 +-
.../org/apache/inlong/agent/plugin/Trigger.java | 9 +-
.../org/apache/inlong/agent/plugin/Validator.java | 4 +-
.../inlong/agent/pojo/DbCollectorTaskResult.java | 5 ++
.../apache/inlong/agent/pojo/JobProfileDto.java | 3 +
.../apache/inlong/agent/pojo/SqlJobProfileDto.java | 3 +
.../inlong/agent/state/AbstractStateWrapper.java | 9 +-
.../inlong/agent/state/StateTransferException.java | 3 +
.../apache/inlong/agent/utils/AgentDbUtils.java | 30 ++++---
.../org/apache/inlong/agent/utils/AgentUtils.java | 79 +++++++++---------
.../org/apache/inlong/agent/utils/ByteUtil.java | 25 ++----
.../org/apache/inlong/agent/utils/ConfigUtil.java | 6 ++
.../org/apache/inlong/agent/utils/ExcuteLinux.java | 11 ++-
.../org/apache/inlong/agent/utils/HttpManager.java | 16 +++-
.../org/apache/inlong/agent/core/AgentMain.java | 10 +--
.../org/apache/inlong/agent/core/AgentManager.java | 16 ++--
.../apache/inlong/agent/core/HeartbeatManager.java | 3 +
.../apache/inlong/agent/core/conf/ConfigJetty.java | 27 ++++--
.../inlong/agent/core/conf/ConfigServlet.java | 32 +++++---
.../java/org/apache/inlong/agent/core/job/Job.java | 18 ++--
.../apache/inlong/agent/core/job/JobManager.java | 37 +++++----
.../apache/inlong/agent/core/job/JobWrapper.java | 20 ++---
.../apache/inlong/agent/core/task/TaskManager.java | 46 +++++------
.../agent/core/task/TaskPositionManager.java | 15 ++--
.../apache/inlong/agent/core/task/TaskWrapper.java | 28 +++----
.../inlong/agent/core/trigger/TriggerManager.java | 46 ++++++-----
.../inlong/agent/plugin/channel/MemoryChannel.java | 37 ++++-----
.../agent/plugin/fetcher/ManagerFetcher.java | 11 ++-
.../plugin/fetcher/ManagerResultFormatter.java | 8 +-
.../agent/plugin/filter/DateFormatRegex.java | 28 ++++---
.../agent/plugin/message/PackProxyMessage.java | 35 ++++----
.../inlong/agent/plugin/message/SequentialID.java | 16 ++--
.../inlong/agent/plugin/sinks/AbstractSink.java | 30 ++++---
.../inlong/agent/plugin/sinks/ProxySink.java | 58 +++++++------
.../inlong/agent/plugin/sinks/SenderManager.java | 78 +++++++++---------
.../inlong/agent/plugin/sources/BinlogSource.java | 19 +++--
.../inlong/agent/plugin/sources/KafkaSource.java | 39 +++++----
.../plugin/sources/reader/AbstractReader.java | 8 ++
.../agent/plugin/sources/reader/BinlogReader.java | 96 +++++++++++-----------
.../agent/plugin/sources/reader/KafkaReader.java | 50 ++++++-----
.../plugin/sources/reader/TextFileReader.java | 34 ++++----
.../sources/snapshot/BinlogSnapshotBase.java | 17 +++-
.../agent/plugin/trigger/DirectoryTrigger.java | 46 ++++++-----
.../inlong/agent/plugin/trigger/PathPattern.java | 27 ++----
.../agent/plugin/utils/BinlogTimeConverter.java | 11 ++-
.../inlong/agent/plugin/utils/PluginUtils.java | 61 ++++++++------
.../agent/plugin/validator/PatternValidator.java | 7 +-
.../agent/plugin/validator/PrefixValidator.java | 7 +-
84 files changed, 927 insertions(+), 752 deletions(-)
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/cache/LocalFileCache.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/cache/LocalFileCache.java
index f6ccacc09..d37901fcd 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/cache/LocalFileCache.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/cache/LocalFileCache.java
@@ -17,14 +17,15 @@
package org.apache.inlong.agent.cache;
+import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.commons.io.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* use local file as cache.
@@ -38,8 +39,8 @@ public class LocalFileCache {
/**
* file with cache time in ms
- * @param file - file
- * @param cacheTime - cache time.
+ * @param file file
+ * @param cacheTime cache time.
*/
public LocalFileCache(File file, long cacheTime) {
this.file = file;
@@ -62,7 +63,6 @@ public class LocalFileCache {
/**
* write string to cache
- * @param info
*/
public void writeToCache(String info) {
lock.writeLock().lock();
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/checkpoint/Checkpoint.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/checkpoint/Checkpoint.java
deleted file mode 100644
index bbdb3f142..000000000
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/checkpoint/Checkpoint.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.checkpoint;
-
-public interface Checkpoint {
-
- /**
- * commit checkpoint.
- */
- void commit();
-
- /**
- * rollback checkpoint.
- */
- void rollback();
-}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AbstractDaemon.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AbstractDaemon.java
index 1affb4e69..653720d14 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AbstractDaemon.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AbstractDaemon.java
@@ -17,6 +17,9 @@
package org.apache.inlong.agent.common;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -24,8 +27,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Providing work threads management, those threads run
@@ -35,11 +36,13 @@ public abstract class AbstractDaemon implements Service {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDaemon.class);
- /** worker thread pool, share it **/
+ /**
+ * worker thread pool, share it
+ **/
private static final ExecutorService WORKER_SERVICES =
- new ThreadPoolExecutor(0, Integer.MAX_VALUE,
- 60L, TimeUnit.SECONDS,
- new SynchronousQueue<Runnable>(), new AgentThreadFactory("AbstractDaemon"));
+ new ThreadPoolExecutor(0, Integer.MAX_VALUE,
+ 60L, TimeUnit.SECONDS,
+ new SynchronousQueue<Runnable>(), new AgentThreadFactory("AbstractDaemon"));
private final List<CompletableFuture<?>> workerFutures;
private boolean runnable = true;
@@ -50,7 +53,7 @@ public abstract class AbstractDaemon implements Service {
/**
* Whether threads can in running state with while loop.
*
- * @return - true if threads can run
+ * @return true if threads can run
*/
public boolean isRunnable() {
return runnable;
@@ -66,7 +69,7 @@ public abstract class AbstractDaemon implements Service {
/**
* Submit work thread to thread pool.
*
- * @param worker - work thread
+ * @param worker work thread
*/
public void submitWorker(Runnable worker) {
CompletableFuture<?> future = CompletableFuture.runAsync(worker, WORKER_SERVICES);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AgentThreadFactory.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AgentThreadFactory.java
index a1bf10045..12c87c584 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AgentThreadFactory.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/AgentThreadFactory.java
@@ -17,11 +17,15 @@
package org.apache.inlong.agent.common;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * AgentThreadFactory, used for creating thread.
+ */
public class AgentThreadFactory implements ThreadFactory {
private static final Logger LOGGER = LoggerFactory.getLogger(AgentThreadFactory.class);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/Service.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/Service.java
index aea5b588b..7ec6ecce7 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/Service.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/common/Service.java
@@ -24,18 +24,21 @@ public interface Service {
/**
* start service
+ *
* @throws Exception
*/
void start() throws Exception;
/**
* stop service
+ *
* @throws Exception
*/
void stop() throws Exception;
/**
* join and wait until getting signal
+ *
* @throws Exception
*/
void join() throws Exception;
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AbstractConfiguration.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AbstractConfiguration.java
index 1a9bb6ce2..37772d9fd 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AbstractConfiguration.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AbstractConfiguration.java
@@ -22,6 +22,11 @@ import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonPrimitive;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.utils.AgentUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
@@ -31,11 +36,10 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.inlong.agent.utils.AgentUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * Base configuration, store and parse config params.
+ */
public abstract class AbstractConfiguration {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConfiguration.class);
@@ -43,7 +47,9 @@ public abstract class AbstractConfiguration {
private final Map<String, JsonPrimitive> configStorage = new HashMap<>();
- /** get config file by class loader **/
+ /**
+ * get config file by class loader
+ **/
private ClassLoader classLoader;
public AbstractConfiguration() {
@@ -54,7 +60,8 @@ public abstract class AbstractConfiguration {
}
/**
- * Check whether all required keys exist
+ * Check whether all required keys exist.
+ *
* @return true if all key exist else false.
*/
public abstract boolean allRequiredKeyExist();
@@ -62,8 +69,8 @@ public abstract class AbstractConfiguration {
/**
* support load config file from json/properties file.
*
- * @param fileName - file name
- * @param isJson - whether is json file
+ * @param fileName file name
+ * @param isJson whether is json file
*/
private void loadResource(String fileName, boolean isJson) {
Reader reader = null;
@@ -78,7 +85,7 @@ public abstract class AbstractConfiguration {
Properties properties = new Properties();
properties.load(reader);
properties.forEach((key, value) -> configStorage.put((String) key,
- new JsonPrimitive((String) value)));
+ new JsonPrimitive((String) value)));
}
}
} catch (Exception ioe) {
@@ -91,7 +98,7 @@ public abstract class AbstractConfiguration {
/**
* load config from json string.
*
- * @param jsonStr - json string
+ * @param jsonStr json string
*/
public void loadJsonStrResource(String jsonStr) {
JsonElement tmpElement = JSON_PARSER.parse(jsonStr);
@@ -101,7 +108,7 @@ public abstract class AbstractConfiguration {
/**
* load config file from CLASS_PATH. config file is json file.
*
- * @param fileName - file name
+ * @param fileName file name
*/
void loadJsonResource(String fileName) {
loadResource(fileName, true);
@@ -114,9 +121,9 @@ public abstract class AbstractConfiguration {
/**
* Convert json string to map
*
- * @param keyDeptPath - map
- * @param dept - json dept
- * @param tmpElement - json element
+ * @param keyDeptPath map
+ * @param dept json dept
+ * @param tmpElement json element
*/
void updateConfig(HashMap<Integer, String> keyDeptPath, int dept, JsonElement tmpElement) {
if (tmpElement instanceof JsonObject) {
@@ -147,8 +154,8 @@ public abstract class AbstractConfiguration {
/**
* get int from config
*
- * @param key - key
- * @param defaultValue - default value
+ * @param key key
+ * @param defaultValue default value
* @return value
*/
public int getInt(String key, int defaultValue) {
@@ -159,7 +166,7 @@ public abstract class AbstractConfiguration {
/**
* get int from config
*
- * @param key - key
+ * @param key key
* @return value
* @throws NullPointerException npe
*/
@@ -174,8 +181,8 @@ public abstract class AbstractConfiguration {
/**
* get long
*
- * @param key - key
- * @param defaultValue - default value
+ * @param key key
+ * @param defaultValue default value
* @return long
*/
public long getLong(String key, long defaultValue) {
@@ -186,8 +193,8 @@ public abstract class AbstractConfiguration {
/**
* get boolean
*
- * @param key - key
- * @param defaultValue - default value
+ * @param key key
+ * @param defaultValue default value
* @return boolean
*/
public boolean getBoolean(String key, boolean defaultValue) {
@@ -198,8 +205,8 @@ public abstract class AbstractConfiguration {
/**
* get string
*
- * @param key - key
- * @param defaultValue - default value
+ * @param key key
+ * @param defaultValue default value
* @return string
*/
public String get(String key, String defaultValue) {
@@ -210,7 +217,7 @@ public abstract class AbstractConfiguration {
/**
* get string or throw npe
*
- * @param key - key
+ * @param key key
* @return string
* @throws NullPointerException if value is null, throw npe
*/
@@ -225,8 +232,8 @@ public abstract class AbstractConfiguration {
/**
* whether key exists
*
- * @param key - key
- * @return - true if key exists else not
+ * @param key key
+ * @return true if key exists else not
*/
public boolean hasKey(String key) {
return configStorage.containsKey(key);
@@ -235,8 +242,8 @@ public abstract class AbstractConfiguration {
/**
* set key/value
*
- * @param key - key
- * @param value - value
+ * @param key key
+ * @param value value
*/
public void set(String key, String value) {
configStorage.put(key, new JsonPrimitive(value));
@@ -258,6 +265,9 @@ public abstract class AbstractConfiguration {
return configStorage;
}
+ /**
+ * get configStorage list, item format: "key=value"
+ */
List<String> getStorageList() {
List<String> result = new ArrayList<>();
for (Map.Entry<String, JsonPrimitive> entry : configStorage.entrySet()) {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AgentConfiguration.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AgentConfiguration.java
index 847011a7e..6203f4a17 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AgentConfiguration.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/AgentConfiguration.java
@@ -17,6 +17,11 @@
package org.apache.inlong.agent.conf;
+import org.apache.commons.io.FileUtils;
+import org.apache.inlong.agent.constant.AgentConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.File;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
@@ -24,10 +29,6 @@ import java.util.Date;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.commons.io.FileUtils;
-import org.apache.inlong.agent.constant.AgentConstants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* agent configuration. Only one instance in the process.
@@ -43,13 +44,12 @@ public class AgentConfiguration extends AbstractConfiguration {
private static final ArrayList<String> LOCAL_RESOURCES = new ArrayList<>();
private static final ReadWriteLock LOCK = new ReentrantReadWriteLock();
+ private static volatile AgentConfiguration agentConf = null;
static {
LOCAL_RESOURCES.add(DEFAULT_CONFIG_FILE);
}
- private static volatile AgentConfiguration agentConf = null;
-
/**
* load config from agent file.
*/
@@ -61,7 +61,8 @@ public class AgentConfiguration extends AbstractConfiguration {
/**
* singleton for agent configuration.
- * @return - static instance of AgentConfiguration
+ *
+ * @return static instance of AgentConfiguration
*/
public static AgentConfiguration getAgentConf() {
if (agentConf == null) {
@@ -88,7 +89,7 @@ public class AgentConfiguration extends AbstractConfiguration {
// TODO: flush to local file as properties file.
try {
String agentConfParent = get(
- AgentConstants.AGENT_CONF_PARENT, AgentConstants.DEFAULT_AGENT_CONF_PARENT);
+ AgentConstants.AGENT_CONF_PARENT, AgentConstants.DEFAULT_AGENT_CONF_PARENT);
File sourceFile = new File(agentConfParent, DEFAULT_CONFIG_FILE);
File targetFile = new File(agentConfParent, getNextBackupFileName());
File tmpFile = new File(agentConfParent, TMP_CONFIG_FILE);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/JobProfile.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/JobProfile.java
index 3d9d24ba2..ffb07d224 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/JobProfile.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/JobProfile.java
@@ -22,7 +22,6 @@ import org.apache.inlong.agent.constant.JobConstants;
/**
* job profile which contains details describing properties of one job.
- *
*/
public class JobProfile extends AbstractConfiguration {
@@ -31,7 +30,6 @@ public class JobProfile extends AbstractConfiguration {
/**
* parse json string to configuration instance.
*
- * @param jsonStr
* @return job configuration
*/
public static JobProfile parseJsonStr(String jsonStr) {
@@ -43,7 +41,7 @@ public class JobProfile extends AbstractConfiguration {
/**
* parse properties file
*
- * @param fileName - file name.
+ * @param fileName file name.
* @return jobConfiguration.
*/
public static JobProfile parsePropertiesFile(String fileName) {
@@ -53,8 +51,9 @@ public class JobProfile extends AbstractConfiguration {
}
/**
- * pase json file.
- * @param fileName - json file name.
+ * parse json file.
+ *
+ * @param fileName json file name.
* @return jobConfiguration.
*/
public static JobProfile parseJsonFile(String fileName) {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/ProfileFetcher.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/ProfileFetcher.java
index 59f31116b..fc676c100 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/ProfileFetcher.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/ProfileFetcher.java
@@ -17,9 +17,10 @@
package org.apache.inlong.agent.conf;
-import java.util.List;
import org.apache.inlong.agent.common.Service;
+import java.util.List;
+
/**
* fetch profile from other system, communicate with json format string
*/
@@ -27,13 +28,15 @@ public interface ProfileFetcher extends Service {
/**
* get job profiles
- * @return - job profile list
+ *
+ * @return job profile list
*/
List<JobProfile> getJobProfiles();
/**
* get trigger profiles
- * @return - trigger profile lisy
+ *
+ * @return trigger profile lisy
*/
- List<TriggerProfile> getTriggerProfiles();
+ List<TriggerProfile> getTriggerProfiles();
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/AgentConstants.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/AgentConstants.java
index f589e5ca3..e92f6e369 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/AgentConstants.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/AgentConstants.java
@@ -19,6 +19,9 @@ package org.apache.inlong.agent.constant;
import org.apache.inlong.agent.utils.AgentUtils;
+/**
+ * Configuration constants of agent.
+ */
public class AgentConstants {
public static final String AGENT_HOME = "agent.home";
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
index c61ed30d2..1d0860b8b 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
@@ -19,6 +19,9 @@ package org.apache.inlong.agent.constant;
import org.apache.inlong.agent.utils.AgentUtils;
+/**
+ * Common constants.
+ */
public class CommonConstants {
public static final String PROXY_NET_TAG = "proxy.net.tag";
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/Db.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/Db.java
index 0d9f15edd..8f5d3ea2d 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/Db.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/Db.java
@@ -19,9 +19,9 @@ package org.apache.inlong.agent.db;
import org.apache.inlong.common.db.CommandEntity;
+import javax.management.openmbean.KeyAlreadyExistsException;
import java.io.Closeable;
import java.util.List;
-import javax.management.openmbean.KeyAlreadyExistsException;
/**
* local storage for key/value.
@@ -32,22 +32,18 @@ public interface Db extends Closeable {
/**
* get command by command id
- * @param commandId
- * @return
*/
CommandEntity getCommand(String commandId);
/**
* put command entity in db
- * @param entity
- * @return
*/
CommandEntity putCommand(CommandEntity entity);
/**
* store keyValue, if key has exists, throw exception.
*
- * @param entity - key/value
+ * @param entity key/value
* @throws NullPointerException key should not be null
* @throws KeyAlreadyExistsException key already exists
*/
@@ -56,7 +52,7 @@ public interface Db extends Closeable {
/**
* store keyValue, if key has exists, overwrite it.
*
- * @param entity - key/value
+ * @param entity key/value
* @return null or old value which is overwritten.
* @throws NullPointerException key should not be null.
*/
@@ -65,7 +61,7 @@ public interface Db extends Closeable {
/**
* remove keyValue by key.
*
- * @param key - key
+ * @param key key
* @return key/value
* @throws NullPointerException key should not be null.
*/
@@ -74,7 +70,7 @@ public interface Db extends Closeable {
/**
* search keyValue list by search key.
*
- * @param searchKey - search keys.
+ * @param searchKey search keys.
* @return key/value list
* @throws NullPointerException search key should not be null.
*/
@@ -82,28 +78,26 @@ public interface Db extends Closeable {
/**
* search commands using ack status
- * @param isAcked
- * @return
*/
List<CommandEntity> searchCommands(boolean isAcked);
/**
* search one keyValue by search key
- * @param searchKey - search key
+ *
+ * @param searchKey search key
* @return null or keyValue
*/
KeyValueEntity searchOne(StateSearchKey searchKey);
/**
* search one keyValue by fileName
- * @param fileName
- * @return
*/
KeyValueEntity searchOne(String fileName);
/**
* find all by prefix key.
- * @param prefix - prefix string
+ *
+ * @param prefix prefix string
* @return list of k/v
*/
List<KeyValueEntity> findAll(String prefix);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/JobProfileDb.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/JobProfileDb.java
index 93769ecac..760abfb6e 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/JobProfileDb.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/JobProfileDb.java
@@ -18,17 +18,19 @@
package org.apache.inlong.agent.db;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.constant.JobConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* Wrapper for job conf persistence.
*/
public class JobProfileDb {
+
private static final Logger LOGGER = LoggerFactory.getLogger(JobProfileDb.class);
private final Db db;
@@ -36,6 +38,9 @@ public class JobProfileDb {
this.db = db;
}
+ /**
+ * get all restart jobs from db
+ */
public List<JobProfile> getRestartJobs() {
List<JobProfile> jobsByState = getJobsByState(StateSearchKey.ACCEPTED);
jobsByState.addAll(getJobsByState(StateSearchKey.RUNNING));
@@ -45,8 +50,9 @@ public class JobProfileDb {
/**
* update job state and search it by key name
- * @param jobInstanceId - job key name
- * @param stateSearchKey - job state
+ *
+ * @param jobInstanceId job key name
+ * @param stateSearchKey job state
*/
public void updateJobState(String jobInstanceId, StateSearchKey stateSearchKey) {
KeyValueEntity entity = db.get(jobInstanceId);
@@ -58,14 +64,15 @@ public class JobProfileDb {
/**
* store job profile
- * @param jobProfile - job profile
+ *
+ * @param jobProfile job profile
*/
public void storeJobFirstTime(JobProfile jobProfile) {
if (jobProfile.allRequiredKeyExist()) {
String keyName = jobProfile.get(JobConstants.JOB_INSTANCE_ID);
jobProfile.setLong(JobConstants.JOB_STORE_TIME, System.currentTimeMillis());
KeyValueEntity entity = new KeyValueEntity(keyName,
- jobProfile.toJsonStr(), jobProfile.get(JobConstants.JOB_DIR_FILTER_PATTERN, ""));
+ jobProfile.toJsonStr(), jobProfile.get(JobConstants.JOB_DIR_FILTER_PATTERN, ""));
entity.setStateSearchKey(StateSearchKey.ACCEPTED);
LOGGER.info("store job {} to db", jobProfile.toJsonStr());
db.put(entity);
@@ -74,6 +81,7 @@ public class JobProfileDb {
/**
* update job profile
+ *
* @param jobProfile
*/
public void updateJobProfile(JobProfile jobProfile) {
@@ -89,8 +97,6 @@ public class JobProfileDb {
/**
* check whether job is finished, note that non-exist job is regarded as finished.
- * @param jobProfile
- * @return
*/
public boolean checkJobfinished(JobProfile jobProfile) {
KeyValueEntity entity = db.get(jobProfile.getInstanceId());
@@ -101,10 +107,19 @@ public class JobProfileDb {
return entity.checkFinished();
}
+ /**
+ * delete job by keyName
+ */
public void deleteJob(String keyName) {
db.remove(keyName);
}
+ /**
+ * get job profile by jobId
+ *
+ * @param jobId jobId
+ * @return JobProfile
+ */
public JobProfile getJobById(String jobId) {
KeyValueEntity keyValueEntity = db.get(jobId);
if (keyValueEntity != null) {
@@ -113,6 +128,11 @@ public class JobProfileDb {
return null;
}
+ /**
+ * remove all expired jobs from db, including success and failed
+ *
+ * @param expireTime expireTime
+ */
public void removeExpireJobs(long expireTime) {
// remove finished tasks
List<KeyValueEntity> successEntityList = db.search(StateSearchKey.SUCCESS);
@@ -126,7 +146,7 @@ public class JobProfileDb {
long currentTime = System.currentTimeMillis();
if (storeTime == 0 || currentTime - storeTime > expireTime) {
LOGGER.info("delete job {} because of timeout store time: {}, expire time: {}",
- entity.getKey(), storeTime, expireTime);
+ entity.getKey(), storeTime, expireTime);
deleteJob(entity.getKey());
}
}
@@ -135,8 +155,9 @@ public class JobProfileDb {
/**
* get job conf by state
- * @param stateSearchKey - state index for searching.
- * @return
+ *
+ * @param stateSearchKey state index for searching.
+ * @return JobProfile
*/
public JobProfile getJob(StateSearchKey stateSearchKey) {
KeyValueEntity entity = db.searchOne(stateSearchKey);
@@ -148,8 +169,6 @@ public class JobProfileDb {
/**
* get job reading specific file
- * @param fileName
- * @return
*/
public JobProfile getJobByFileName(String fileName) {
KeyValueEntity entity = db.searchOne(fileName);
@@ -161,8 +180,9 @@ public class JobProfileDb {
/**
* get list of job profiles.
- * @param stateSearchKey - state search key.
- * @return - list of job profile.
+ *
+ * @param stateSearchKey state search key.
+ * @return list of job profile.
*/
public List<JobProfile> getJobsByState(StateSearchKey stateSearchKey) {
List<KeyValueEntity> entityList = db.search(stateSearchKey);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/KeyValueEntity.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/KeyValueEntity.java
index 8c65599b1..5cbda1e4f 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/KeyValueEntity.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/KeyValueEntity.java
@@ -75,6 +75,7 @@ public class KeyValueEntity {
/**
* convert keyValue to job profile
+ *
* @return JobConfiguration
*/
public JobProfile getAsJobProfile() {
@@ -84,7 +85,6 @@ public class KeyValueEntity {
/**
* convert keyValue to trigger profile
- * @return
*/
public TriggerProfile getAsTriggerProfile() {
return TriggerProfile.parseJsonStr(getJsonValue());
@@ -92,7 +92,6 @@ public class KeyValueEntity {
/**
* check whether the entity is finished
- * @return
*/
public boolean checkFinished() {
return stateSearchKey.equals(StateSearchKey.SUCCESS)
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/LocalProfile.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/LocalProfile.java
index da4cba0e2..923636c37 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/LocalProfile.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/LocalProfile.java
@@ -17,6 +17,10 @@
package org.apache.inlong.agent.db;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -24,14 +28,12 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Stream;
-import org.apache.inlong.agent.conf.JobProfile;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* profile from local file
*/
public class LocalProfile {
+
private static final Logger LOGGER = LoggerFactory.getLogger(LocalProfile.class);
private static final String JSON_SUFFIX = ".json";
@@ -45,6 +47,11 @@ public class LocalProfile {
this.filePath = Paths.get(fileName);
}
+ /**
+ * If filePath is a directory, read all local file and construct job profiles
+ *
+ * @return list of JobProfiles
+ */
public List<JobProfile> readFromLocal() {
List<JobProfile> profileList = new ArrayList<>();
try {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/RocksDbImp.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/RocksDbImp.java
index c8330f46d..7ec0fb02d 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/RocksDbImp.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/RocksDbImp.java
@@ -17,16 +17,7 @@
package org.apache.inlong.agent.db;
-import static java.util.Objects.requireNonNull;
-
import com.google.gson.Gson;
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.constant.AgentConstants;
import org.apache.inlong.common.db.CommandEntity;
@@ -43,6 +34,16 @@ import org.rocksdb.Statistics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import static java.util.Objects.requireNonNull;
+
/**
* DB implement based on rocks db.
*/
@@ -128,6 +129,9 @@ public class RocksDbImp implements Db {
return new ColumnFamilyDescriptor(columnFamilyName, new ColumnFamilyOptions());
}
+ /**
+ * add columnFamilyName
+ */
public void addColumnFamily(String columnFamilyName) {
columnDescriptorMap.computeIfAbsent(columnFamilyName, colFamilyName -> {
try {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/TriggerProfileDb.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/TriggerProfileDb.java
index d317d1672..af35fbe51 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/TriggerProfileDb.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/db/TriggerProfileDb.java
@@ -17,14 +17,15 @@
package org.apache.inlong.agent.db;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.inlong.agent.conf.TriggerProfile;
import org.apache.inlong.agent.constant.CommonConstants;
import org.apache.inlong.agent.constant.JobConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* db interface for trigger profile.
*/
@@ -40,7 +41,8 @@ public class TriggerProfileDb {
/**
* get trigger list from db.
- * @return - list of trigger
+ *
+ * @return list of trigger
*/
public List<TriggerProfile> getTriggers() {
// potential performance issue, needs to find out the speed.
@@ -54,13 +56,14 @@ public class TriggerProfileDb {
/**
* store trigger profile.
- * @param trigger - trigger
+ *
+ * @param trigger trigger
*/
public void storeTrigger(TriggerProfile trigger) {
if (trigger.allRequiredKeyExist()) {
String keyName = CommonConstants.TRIGGER_ID_PREFIX + trigger.get(JobConstants.JOB_ID);
KeyValueEntity entity = new KeyValueEntity(keyName,
- trigger.toJsonStr(), trigger.get(JobConstants.JOB_DIR_FILTER_PATTERN));
+ trigger.toJsonStr(), trigger.get(JobConstants.JOB_DIR_FILTER_PATTERN));
KeyValueEntity oldEntity = db.put(entity);
if (oldEntity != null) {
LOGGER.warn("trigger profile {} has been replaced", oldEntity.getKey());
@@ -70,7 +73,6 @@ public class TriggerProfileDb {
/**
* delete trigger by id.
- * @param id
*/
public void deleteTrigger(String id) {
db.remove(CommonConstants.TRIGGER_ID_PREFIX + id);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/DefaultMessage.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/DefaultMessage.java
index 3c735cffb..fb34a5dc0 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/DefaultMessage.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/DefaultMessage.java
@@ -17,11 +17,16 @@
package org.apache.inlong.agent.message;
+import org.apache.inlong.agent.plugin.Message;
+
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
-import org.apache.inlong.agent.plugin.Message;
+/**
+ * DefaultMessage is used in inner-data transfer, including two parts, header and body. Header is the attributes of
+ * message, and body is the content of message.
+ */
public class DefaultMessage implements Message {
private final byte[] body;
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/ProxyMessage.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/ProxyMessage.java
index c189b823b..f07d8ceba 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/ProxyMessage.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/ProxyMessage.java
@@ -17,13 +17,14 @@
package org.apache.inlong.agent.message;
+import org.apache.inlong.agent.plugin.Message;
+
+import java.util.Map;
+
import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_DATA;
import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_GROUP_ID;
import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_STREAM_ID;
-import java.util.Map;
-import org.apache.inlong.agent.plugin.Message;
-
/**
* Bus message with body, header, inlongGroupId and inlongStreamId.
*/
@@ -49,6 +50,16 @@ public class ProxyMessage implements Message {
this.batchKey = dataKey + inlongStreamId;
}
+ /**
+ * Transform Message to ProxyMessage
+ *
+ * @param message Message
+ * @return ProxyMessage
+ */
+ public static ProxyMessage parse(Message message) {
+ return new ProxyMessage(message.getBody(), message.getHeader());
+ }
+
public String getDataKey() {
return dataKey;
}
@@ -84,8 +95,4 @@ public class ProxyMessage implements Message {
public String getBatchKey() {
return batchKey;
}
-
- public static ProxyMessage parse(Message message) {
- return new ProxyMessage(message.getBody(), message.getHeader());
- }
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricException.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricException.java
index ada7edd66..8129a8c04 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricException.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricException.java
@@ -17,6 +17,9 @@
package org.apache.inlong.agent.metrics;
+/**
+ * MetricException
+ */
public class MetricException extends RuntimeException {
public MetricException(String message, Exception ex) {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricSnapshot.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricSnapshot.java
index bd729c84c..64172d9f6 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricSnapshot.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricSnapshot.java
@@ -17,6 +17,9 @@
package org.apache.inlong.agent.metrics;
+/**
+ * MetricSnapshot Template
+ */
public interface MetricSnapshot<T> {
T snapshot();
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/Metrics.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/Metrics.java
index f3ae8e75c..038296255 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/Metrics.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/Metrics.java
@@ -31,22 +31,16 @@ public @interface Metrics {
/**
* Metrics name
- *
- * @return
*/
String name() default "";
/**
* Metrics context
- *
- * @return
*/
String context() default "";
/**
* Metrics description
- *
- * @return
*/
String desc() default "";
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricsRegister.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricsRegister.java
index 7ad552b75..7124d58b6 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricsRegister.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/MetricsRegister.java
@@ -17,14 +17,6 @@
package org.apache.inlong.agent.metrics;
-import java.lang.annotation.Annotation;
-import java.lang.management.ManagementFactory;
-import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
import org.apache.inlong.agent.metrics.counter.CounterInt;
import org.apache.inlong.agent.metrics.counter.CounterLong;
import org.apache.inlong.agent.metrics.gauge.GaugeInt;
@@ -35,10 +27,20 @@ import org.apache.inlong.agent.utils.AgentUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import java.lang.annotation.Annotation;
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
/**
* register for metrics.
*/
public class MetricsRegister {
+
private static final Logger LOGGER = LoggerFactory.getLogger(MetricsRegister.class);
private static final String DOMAIN_PREFIX = "Agent:";
@@ -54,6 +56,7 @@ public class MetricsRegister {
/**
* register object name for metric
+ *
* @param agentDynamicMBean agent mbean
*/
private static void innerRegister(AgentDynamicMBean agentDynamicMBean) {
@@ -71,6 +74,9 @@ public class MetricsRegister {
}
}
+ /**
+ * register metric for object
+ */
public static void register(String module, String aspect, String desc, Object source) {
List<MetricMeta> metricMetaList = handleFieldAnnotation(source);
MetricsMeta metricsMeta = handleClassAnnotation(source, metricMetaList);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/audit/AuditUtils.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/audit/AuditUtils.java
index 0309d95e3..9198ade54 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/audit/AuditUtils.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/audit/AuditUtils.java
@@ -71,8 +71,6 @@ public class AuditUtils {
/**
* add
- *
- * @param auditID
*/
public static void add(int auditID, String inlongGroupId, String inlongStreamId, long logTime) {
if (!IS_AUDIT) {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/Counter.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/Counter.java
index d035a6918..eef4d74b0 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/Counter.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/Counter.java
@@ -19,6 +19,9 @@ package org.apache.inlong.agent.metrics.counter;
import org.apache.inlong.agent.metrics.MutableMetric;
+/**
+ * Counter
+ */
public interface Counter extends MutableMetric {
/**
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/CounterInt.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/CounterInt.java
index ef1215be0..86b8bc22f 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/CounterInt.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/counter/CounterInt.java
@@ -19,6 +19,9 @@ package org.apache.inlong.agent.metrics.counter;
import java.util.concurrent.atomic.AtomicInteger;
+/**
+ * atomic int for counter
+ */
public class CounterInt implements Counter {
private AtomicInteger value = new AtomicInteger(0);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/Gauge.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/Gauge.java
index 2ee52a20f..26fb02cdc 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/Gauge.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/Gauge.java
@@ -19,8 +19,14 @@ package org.apache.inlong.agent.metrics.gauge;
import org.apache.inlong.agent.metrics.MutableMetric;
+/**
+ * Gauge Template
+ */
public interface Gauge<T> extends MutableMetric {
+ /**
+ * set
+ */
void set(T num);
/**
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeInt.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeInt.java
index 024628977..28b22fa91 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeInt.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeInt.java
@@ -19,6 +19,9 @@ package org.apache.inlong.agent.metrics.gauge;
import java.util.concurrent.atomic.AtomicInteger;
+/**
+ * GaugeInt
+ */
public class GaugeInt implements Gauge<Integer> {
private final AtomicInteger value = new AtomicInteger(0);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeLong.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeLong.java
index 4bf01bf52..3f48ec7c9 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeLong.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/gauge/GaugeLong.java
@@ -19,6 +19,9 @@ package org.apache.inlong.agent.metrics.gauge;
import java.util.concurrent.atomic.AtomicLong;
+/**
+ * GaugeLong
+ */
public class GaugeLong implements Gauge<Long> {
private final AtomicLong value = new AtomicLong(0);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricMeta.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricMeta.java
index adf889017..694bc7ba4 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricMeta.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricMeta.java
@@ -17,13 +17,6 @@
package org.apache.inlong.agent.metrics.meta;
-import static org.apache.inlong.agent.metrics.Metric.Type.COUNTER_INT;
-import static org.apache.inlong.agent.metrics.Metric.Type.COUNTER_LONG;
-import static org.apache.inlong.agent.metrics.Metric.Type.DEFAULT;
-import static org.apache.inlong.agent.metrics.Metric.Type.GAUGE_INT;
-import static org.apache.inlong.agent.metrics.Metric.Type.GAUGE_LONG;
-
-import java.lang.reflect.Field;
import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.agent.metrics.Metric;
import org.apache.inlong.agent.metrics.counter.CounterInt;
@@ -31,6 +24,14 @@ import org.apache.inlong.agent.metrics.counter.CounterLong;
import org.apache.inlong.agent.metrics.gauge.GaugeInt;
import org.apache.inlong.agent.metrics.gauge.GaugeLong;
+import java.lang.reflect.Field;
+
+import static org.apache.inlong.agent.metrics.Metric.Type.COUNTER_INT;
+import static org.apache.inlong.agent.metrics.Metric.Type.COUNTER_LONG;
+import static org.apache.inlong.agent.metrics.Metric.Type.DEFAULT;
+import static org.apache.inlong.agent.metrics.Metric.Type.GAUGE_INT;
+import static org.apache.inlong.agent.metrics.Metric.Type.GAUGE_LONG;
+
/**
* this class is related to {@link Metric}
*/
@@ -41,6 +42,13 @@ public class MetricMeta {
private String desc;
private Field field;
+ /**
+ * build metricMeta based on Metric and Field params
+ *
+ * @param annotation Metric
+ * @param field Field
+ * @return MetricMeta
+ */
public static MetricMeta build(Metric annotation, Field field) {
MetricMeta metricMeta = new MetricMeta();
metricMeta.name = StringUtils.capitalize(field.getName());
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricsMeta.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricsMeta.java
index 03a697ff9..972a6a8fa 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricsMeta.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/meta/MetricsMeta.java
@@ -17,9 +17,10 @@
package org.apache.inlong.agent.metrics.meta;
-import java.util.List;
import org.apache.inlong.agent.metrics.Metrics;
+import java.util.List;
+
/**
* This class is related to {@link Metrics}
*/
@@ -33,6 +34,13 @@ public class MetricsMeta {
private MetricsMeta() {
}
+ /**
+ * build metricsMeta based on Metrics and List<MetricMeta> params
+ *
+ * @param metrics Metrics
+ * @param metricMetaList List<MetricMeta>
+ * @return metricsMeta
+ */
public static MetricsMeta build(Metrics metrics,
List<MetricMeta> metricMetaList) {
MetricsMeta metricsMeta = new MetricsMeta();
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Channel.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Channel.java
index ffda65fc1..08cb50d62 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Channel.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Channel.java
@@ -26,27 +26,16 @@ public interface Channel extends Stage {
/**
* write message
- *
- * @param message - message
*/
void push(Message message);
/**
* write message with timeout
- *
- * @param message
- * @param timeout
- * @param unit
- * @return
*/
boolean push(Message message, long timeout, TimeUnit unit);
/**
* read message with timeout
- *
- * @param timeout
- * @param unit
- * @return
*/
Message pull(long timeout, TimeUnit unit);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/MessageFilter.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/MessageFilter.java
index 966dd771e..3784b5d86 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/MessageFilter.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/MessageFilter.java
@@ -17,14 +17,18 @@
package org.apache.inlong.agent.plugin;
+/**
+ * message filter
+ */
public interface MessageFilter {
/**
* split a message to get stream id string
* used when the file is separated with different steam id
+ *
* @param message the input message
* @param fieldSplitter fieldSplitter used when split a line
- * @return
+ * @return message after filtration
*/
String filterStreamId(Message message, byte[] fieldSplitter);
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Reader.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Reader.java
index 6bd1e522b..55b768c72 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Reader.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Reader.java
@@ -26,19 +26,17 @@ public interface Reader extends Stage {
/**
* Read message
*
- * @return - message
+ * @return message
*/
Message read();
/**
* Whether finish reading
- * @return
*/
boolean isFinished();
/**
* Return the reader's reading source name
- * @return
*/
String getReadSource();
@@ -55,7 +53,6 @@ public interface Reader extends Stage {
/**
* get snapshot of the reader
- * @return
*/
String getSnapshot();
@@ -66,7 +63,6 @@ public interface Reader extends Stage {
/**
* source is exist
- * @return
*/
boolean isSourceExist();
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java
index f9fc7d32c..43eb2323f 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java
@@ -27,12 +27,13 @@ public interface Sink extends Stage {
/**
* Write data into data center
*
- * @param message - message
+ * @param message message
*/
void write(Message message);
/**
* set source file name where the message is generated
+ *
* @param sourceFileName
*/
void setSourceName(String sourceFileName);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Source.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Source.java
index a1da45982..8211343f0 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Source.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Source.java
@@ -17,9 +17,10 @@
package org.apache.inlong.agent.plugin;
-import java.util.List;
import org.apache.inlong.agent.conf.JobProfile;
+import java.util.List;
+
/**
* Source can be split into multiple reader.
*/
@@ -29,7 +30,7 @@ public interface Source {
* Split source into a list of readers.
*
* @param conf job conf
- * @return - list of reader
+ * @return list of reader
*/
List<Reader> split(JobProfile conf);
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Stage.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Stage.java
index 2a9a96c98..8fbaca109 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Stage.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Stage.java
@@ -27,7 +27,7 @@ public interface Stage {
/**
* Init job.
*
- * @param jobConf - job config
+ * @param jobConf job config
*/
void init(JobProfile jobConf);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Trigger.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Trigger.java
index 02160b58d..3757145b8 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Trigger.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Trigger.java
@@ -17,10 +17,11 @@
package org.apache.inlong.agent.plugin;
-import java.io.IOException;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.conf.TriggerProfile;
+import java.io.IOException;
+
/**
* Trigger interface, which generates job in condition.
*/
@@ -28,7 +29,7 @@ public interface Trigger {
/**
* init trigger by trigger profile
- * @param profile
+ *
* @throws IOException
*/
void init(TriggerProfile profile) throws IOException;
@@ -45,13 +46,13 @@ public interface Trigger {
/**
* fetch job profile from trigger
- * @return - job profile
+ *
+ * @return job profile
*/
JobProfile fetchJobProfile();
/**
* get trigger profile
- * @return
*/
TriggerProfile getTriggerProfile();
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Validator.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Validator.java
index 6ee626887..157c2b2ef 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Validator.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Validator.java
@@ -23,9 +23,7 @@ package org.apache.inlong.agent.plugin;
public interface Validator {
/**
- * validate
- * @param messageLine
- * @return
+ * validate whether messageLine meets specified requirement
*/
boolean validate(String messageLine);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/DbCollectorTaskResult.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/DbCollectorTaskResult.java
index 0c1b396de..320153c0a 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/DbCollectorTaskResult.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/DbCollectorTaskResult.java
@@ -30,6 +30,11 @@ public class DbCollectorTaskResult {
private DbCollectorTask data = new DbCollectorTask();
+ /**
+ * get JobProfile based on DbCollectorTask
+ *
+ * @return JobProfile
+ */
public JobProfile getJobProfile() {
JobProfile profile = null;
if (data == null) {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
index 9c187f57a..55a5a7815 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
@@ -155,6 +155,9 @@ public class JobProfileDto {
return proxy;
}
+ /**
+ * convert DataConfig to TriggerProfile
+ */
public static TriggerProfile convertToTriggerProfile(DataConfig dataConfigs) {
if (!dataConfigs.isValid()) {
throw new IllegalArgumentException("input dataConfig" + dataConfigs + "is invalid please check");
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/SqlJobProfileDto.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/SqlJobProfileDto.java
index 6293de693..860588789 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/SqlJobProfileDto.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/SqlJobProfileDto.java
@@ -71,6 +71,9 @@ public class SqlJobProfileDto {
return proxy;
}
+ /**
+ * convert DbCollectorTask to JobProfile
+ */
public static JobProfile convertToJobProfile(DbCollectorTask task) {
if (!task.isValid()) {
throw new IllegalArgumentException("input task" + task + "is invalid please check");
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/AbstractStateWrapper.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/AbstractStateWrapper.java
index aa51fe80e..640df68bc 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/AbstractStateWrapper.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/AbstractStateWrapper.java
@@ -17,13 +17,14 @@
package org.apache.inlong.agent.state;
-import java.util.HashMap;
-import java.util.Map;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.Map;
+
public abstract class AbstractStateWrapper implements Runnable {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractStateWrapper.class);
@@ -48,7 +49,7 @@ public abstract class AbstractStateWrapper implements Runnable {
/**
* change state and execute callback functions
*
- * @param nextState - next state
+ * @param nextState next state
*/
public synchronized void doChangeState(State nextState) {
LOGGER.debug("state change, current state is {}, next state is {}", currentState, nextState);
@@ -63,8 +64,6 @@ public abstract class AbstractStateWrapper implements Runnable {
/**
* determine the exception
- *
- * @return
*/
public boolean isException() {
State tmpState = currentState;
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/StateTransferException.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/StateTransferException.java
index 34ed0e3bc..6972f28e9 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/StateTransferException.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/state/StateTransferException.java
@@ -17,6 +17,9 @@
package org.apache.inlong.agent.state;
+/**
+ * StateTransferException
+ */
public class StateTransferException extends RuntimeException {
public StateTransferException(State begin, State end) {
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentDbUtils.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentDbUtils.java
index e86631844..87ffe7b1f 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentDbUtils.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentDbUtils.java
@@ -17,6 +17,9 @@
package org.apache.inlong.agent.utils;
+import org.apache.commons.dbutils.DbUtils;
+import org.apache.commons.lang3.StringUtils;
+
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
@@ -24,20 +27,21 @@ import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.apache.commons.dbutils.DbUtils;
-import org.apache.commons.lang3.StringUtils;
+/**
+ * AgentDbUtils
+ */
public class AgentDbUtils {
private static final Pattern PATTERN =
- Pattern.compile("\\$\\{(((0x)|(0X)|o|O)??[0-9a-fA-F]+?) *, "
- + "*(((0x)|(0X)|o|O)??[0-9a-fA-F]+?) *(, *[0-9]*?)??}");
+ Pattern.compile("\\$\\{(((0x)|(0X)|o|O)??[0-9a-fA-F]+?) *, "
+ + "*(((0x)|(0X)|o|O)??[0-9a-fA-F]+?) *(, *[0-9]*?)??}");
private static final Pattern OCT_PATTERN = Pattern.compile("^o[0-7]+?$");
private static final Pattern DEC_PATTERN = Pattern.compile("^[0-9]+?$");
private static final int HEX_MODE = 16;
private static final int EIGHT_MODE = 8;
private static final String HEX_PREFIX = "0x";
- private static final String O_PREFIX = "o";
+ private static final String O_PREFIX = "o";
/**
* Attempts to establish a connection to the database from the XML configurations. If failed,
@@ -136,9 +140,9 @@ public class AgentDbUtils {
* Transfer string pattern into a list of real string.
* For example: ${1, 99} = 1, 2, 3, ... 98,
* 99 <br> ${01, 99} = 01, 02, ... 98, 99 <br>
- * ${0x0,0xff} = 1, 2, ... fe, ff <br> ${0x00,0xff}
+ * ${0x0,0xff} = 1, 2, ... fe, ff <br> ${0x00,0xff}
* = 01, 02, ... fe, ff <br> ${O1,O10} = 1, 2,... 7, 10<br>
- * ${O01,O10} = 01, 02,... 07, 10<br>
+ * ${O01,O10} = 01, 02,... 07, 10<br>
*
* test_${0x00,0x12,5} = test_00, test_05, test_0a, test_0f<br>
*
@@ -214,10 +218,10 @@ public class AgentDbUtils {
}
private static ArrayList<String>[] formatStartNum(ArrayList<String> startNum,
- ArrayList<String> endNum,
- ArrayList<Integer> modes,
- ArrayList<Integer> steps,
- StringBuffer sb) {
+ ArrayList<String> endNum,
+ ArrayList<Integer> modes,
+ ArrayList<Integer> steps,
+ StringBuffer sb) {
@SuppressWarnings("unchecked")
ArrayList<String>[] tempArray = new ArrayList[startNum.size() + 1];
tempArray[0] = new ArrayList<String>();
@@ -232,8 +236,8 @@ public class AgentDbUtils {
for (String currentPath : tempArray[index]) {
for (int i = parseInt(start); i <= parseInt(end); i = i + step) {
tempArray[index + 1].add(currentPath.replaceAll(
- "\\$\\{" + index + "}",
- format(i, lengthEquals, end.length(), mode)));
+ "\\$\\{" + index + "}",
+ format(i, lengthEquals, end.length(), mode)));
}
}
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentUtils.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentUtils.java
index e02540149..9f5e7aec7 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentUtils.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/AgentUtils.java
@@ -17,6 +17,13 @@
package org.apache.inlong.agent.utils;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.inlong.agent.conf.AgentConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.Closeable;
import java.io.File;
import java.io.InputStream;
@@ -44,12 +51,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.inlong.agent.conf.AgentConfiguration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import static org.apache.inlong.agent.constant.AgentConstants.AGENT_LOCAL_IP;
import static org.apache.inlong.agent.constant.AgentConstants.AGENT_LOCAL_UUID;
@@ -57,11 +58,11 @@ import static org.apache.inlong.agent.constant.AgentConstants.AGENT_LOCAL_UUID_O
import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_AGENT_LOCAL_UUID_OPEN;
import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_LOCAL_IP;
+/**
+ * AgentUtils
+ */
public class AgentUtils {
- private static final Logger LOGGER = LoggerFactory.getLogger(AgentUtils.class);
- private static final AtomicLong INDEX = new AtomicLong(0);
- private static final String HEX_PREFIX = "0x";
public static final String EQUAL = "=";
public static final String M_VALUE = "m";
public static final String ADDITION_SPLITTER = "&";
@@ -73,10 +74,14 @@ public class AgentUtils {
public static final String HOUR = "H";
public static final String HOUR_LOW_CASE = "h";
public static final String MINUTE = "m";
+ private static final Logger LOGGER = LoggerFactory.getLogger(AgentUtils.class);
+ private static final AtomicLong INDEX = new AtomicLong(0);
+ private static final String HEX_PREFIX = "0x";
/**
* get md5 of file.
- * @param file - file name
+ *
+ * @param file file name
* @return
*/
public static String getFileMd5(File file) {
@@ -90,7 +95,6 @@ public class AgentUtils {
/**
* return system current time
- * @return
*/
public static long getCurrentTime() {
return System.currentTimeMillis();
@@ -99,7 +103,7 @@ public class AgentUtils {
/**
* finally close resources
*
- * @param resource - resource which is closable.
+ * @param resource resource which is closable.
*/
public static void finallyClose(Closeable resource) {
if (resource != null) {
@@ -114,7 +118,7 @@ public class AgentUtils {
/**
* finally close resources.
*
- * @param resource - resource which is closable.
+ * @param resource resource which is closable.
*/
public static void finallyClose(AutoCloseable resource) {
if (resource != null) {
@@ -142,7 +146,7 @@ public class AgentUtils {
/**
* Get declare methods.
*
- * @param clazz - class of field from method return
+ * @param clazz class of field from method return
* @return list of methods
*/
public static List<Method> getDeclaredMethodsIncludingInherited(Class<?> clazz) {
@@ -156,8 +160,6 @@ public class AgentUtils {
/**
* get random int of [seed, seed * 2]
- * @param seed
- * @return
*/
public static int getRandomBySeed(int seed) {
return ThreadLocalRandom.current().nextInt(0, seed) + seed;
@@ -194,15 +196,19 @@ public class AgentUtils {
/**
* Get uniq id with timestamp and index.
- * @param id - job id
- * @param index - job index
+ *
+ * @param id job id
+ * @param index job index
* @return uniq id
*/
public static String getUniqId(String prefix, String id, long index) {
long currentTime = System.currentTimeMillis() / 1000;
- return prefix + currentTime + "_" + id + "_" + index;
+ return prefix + currentTime + "_" + id + "_" + index;
}
+ /**
+ * sleep millisecond
+ */
public static void silenceSleepInMs(long millisecond) {
try {
TimeUnit.MILLISECONDS.sleep(millisecond);
@@ -211,6 +217,9 @@ public class AgentUtils {
}
}
+ /**
+ * sleep minutes
+ */
public static void silenceSleepInMinute(long minutes) {
try {
TimeUnit.MINUTES.sleep(minutes);
@@ -232,13 +241,14 @@ public class AgentUtils {
/**
* formatter for current time
- * @param formatter
- * @return
*/
public static String formatCurrentTime(String formatter) {
return formatCurrentTime(formatter, Locale.getDefault());
}
+ /**
+ * formatter for current time based on zone
+ */
public static String formatCurrentTime(String formatter, Locale locale) {
ZonedDateTime zoned = ZonedDateTime.now();
// TODO: locale seems not working
@@ -247,10 +257,11 @@ public class AgentUtils {
/**
* formatter with time offset
- * @param formatter - formatter string
- * @param day - day offset
- * @param hour - hour offset
- * @param min - min offset
+ *
+ * @param formatter formatter string
+ * @param day day offset
+ * @param hour hour offset
+ * @param min min offset
* @return current time with offset
*/
public static String formatCurrentTimeWithOffset(String formatter, int day, int hour, int min) {
@@ -266,8 +277,8 @@ public class AgentUtils {
/**
* whether all class of path name are matched
*
- * @param pathStr - path string
- * @param patternStr - regex pattern
+ * @param pathStr path string
+ * @param patternStr regex pattern
* @return true if all match
*/
public static boolean regexMatch(String pathStr, String patternStr) {
@@ -287,8 +298,6 @@ public class AgentUtils {
/**
* parse addition attr, the attributes must be send in proxy sender
- * @param additionStr
- * @return
*/
public static Pair<String, Map<String, String>> parseAddAttr(String additionStr) {
Map<String, String> attr = new HashMap<>();
@@ -310,9 +319,6 @@ public class AgentUtils {
/**
* the attrs in pairs can be complicated in online env
- * @param attr
- * @param s
- * @param pairs
*/
private static void getAttrs(Map<String, String> attr, String s, String[] pairs) {
// when addiction attr be like "m=10&__addcol1__worldid="
@@ -325,8 +331,6 @@ public class AgentUtils {
/**
* get addition attributes in additionStr
- * @param additionStr
- * @return
*/
public static Map<String, String> getAdditionAttr(String additionStr) {
Pair<String, Map<String, String>> mValueAttrs = parseAddAttr(additionStr);
@@ -335,8 +339,6 @@ public class AgentUtils {
/**
* get m value in additionStr
- * @param addictiveAttr
- * @return
*/
public static String getmValue(String addictiveAttr) {
Pair<String, Map<String, String>> mValueAttrs = parseAddAttr(addictiveAttr);
@@ -356,7 +358,7 @@ public class AgentUtils {
public static String fetchLocalUuid() {
String uuid = "";
if (!AgentConfiguration.getAgentConf()
- .getBoolean(AGENT_LOCAL_UUID_OPEN, DEFAULT_AGENT_LOCAL_UUID_OPEN)) {
+ .getBoolean(AGENT_LOCAL_UUID_OPEN, DEFAULT_AGENT_LOCAL_UUID_OPEN)) {
return uuid;
}
try {
@@ -379,9 +381,6 @@ public class AgentUtils {
/**
* time str convert to mill sec
- * @param time
- * @param cycleUnit
- * @return
*/
public static long timeStrConvertToMillSec(String time, String cycleUnit) {
long defaultTime = System.currentTimeMillis();
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ByteUtil.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ByteUtil.java
index 7a52b6b4f..9c129c5e9 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ByteUtil.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ByteUtil.java
@@ -20,16 +20,16 @@ package org.apache.inlong.agent.utils;
import java.util.ArrayList;
import java.util.List;
+/**
+ * ByteUtil
+ */
public class ByteUtil {
+
/**
* Splits the source array into multiple array segments using the given separator, up to a
* maximum of count items. This will naturally produce copied byte arrays for each of the split
* segments. To identify the split ranges without the array copies, see {@link
* ByteUtil#splitRanges(byte[], byte[])}.
- *
- * @param source
- * @param separator
- * @return
*/
public static byte[][] split(byte[] source, byte[] separator) {
return split(source, separator, -1);
@@ -40,10 +40,6 @@ public class ByteUtil {
* maximum of count items. This will naturally produce copied byte arrays for each of the split
* segments. To identify the split ranges without the array copies, see {@link
* ByteUtil#splitRanges(byte[], byte[])}.
- *
- * @param source
- * @param separator
- * @return
*/
public static byte[][] split(byte[] source, byte[] separator, int limit) {
List<Range> segments = splitRanges(source, separator, limit);
@@ -107,10 +103,6 @@ public class ByteUtil {
/**
* Returns a single byte array containing all of the individual component arrays separated by
* the separator array.
- *
- * @param separator
- * @param components
- * @return
*/
public static byte[] join(byte[] separator, byte[]... components) {
if (components == null || components.length == 0) {
@@ -184,11 +176,6 @@ public class ByteUtil {
/**
* Returns a copy of the source byte array, starting at offset for the given length. If the
* offset + length is out of bounds for the array, returns null.
- *
- * @param source
- * @param offset
- * @param length
- * @return
*/
public static byte[] safeCopy(byte[] source, int offset, int length) {
if (length < 0 || source.length < offset + length) {
@@ -213,8 +200,8 @@ public class ByteUtil {
public Range(int start, int end) {
if (start < 0 || end < start) {
throw new IllegalArgumentException(
- "Invalid range, required that: 0 <= start <= end; start=" + start
- + ", end=" + end);
+ "Invalid range, required that: 0 <= start <= end; start=" + start
+ + ", end=" + end);
}
this.startIdx = start;
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ConfigUtil.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ConfigUtil.java
index 5dd1dcdd3..fd6fad2b4 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ConfigUtil.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ConfigUtil.java
@@ -24,8 +24,14 @@ import org.apache.inlong.agent.conf.AgentConfiguration;
import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_PROMETHEUS_ENABLE;
import static org.apache.inlong.agent.constant.AgentConstants.PROMETHEUS_ENABLE;
+/**
+ * ConfigUtil
+ */
public class ConfigUtil {
+ /**
+ * whether use prometheus metrics
+ */
public static boolean isPrometheusEnabled() {
AgentConfiguration conf = AgentConfiguration.getAgentConf();
return conf.getBoolean(PROMETHEUS_ENABLE, DEFAULT_PROMETHEUS_ENABLE);
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ExcuteLinux.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ExcuteLinux.java
index cc8f508e0..8d9c0e23a 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ExcuteLinux.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/ExcuteLinux.java
@@ -20,13 +20,22 @@ package org.apache.inlong.agent.utils;
import java.io.BufferedReader;
import java.io.InputStreamReader;
+/**
+ * ExecuteLinux cmd
+ */
public class ExcuteLinux {
+ /**
+ * execute linux cmd
+ *
+ * @param commandStr cmd
+ * @return result of execution
+ */
public static String exeCmd(String commandStr) {
String result = null;
try {
- String[] cmd = new String[]{"/bin/sh", "-c",commandStr};
+ String[] cmd = new String[]{"/bin/sh", "-c", commandStr};
Process ps = Runtime.getRuntime().exec(cmd);
BufferedReader br = new BufferedReader(new InputStreamReader(ps.getInputStream()));
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java
index f31e8a512..d11c509cc 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java
@@ -38,6 +38,9 @@ import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_HTTP_SUCCE
import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_REQUEST_TIMEOUT;
import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_REQUEST_TIMEOUT;
+/**
+ * Perform http operation
+ */
public class HttpManager {
private static final Logger LOGGER = LoggerFactory.getLogger(HttpManager.class);
@@ -58,7 +61,7 @@ public class HttpManager {
/**
* construct http client
*
- * @param timeout - timeout setting
+ * @param timeout timeout setting
* @return closeable timeout
*/
private synchronized CloseableHttpClient constructHttpClient(int timeout) {
@@ -74,6 +77,12 @@ public class HttpManager {
return httpClientBuilder.build();
}
+ /**
+ * doPost
+ *
+ * @param dto content body needed to post
+ * @return response
+ */
public String doSentPost(String url, Object dto) {
try {
HttpPost post = getHttpPost(url);
@@ -100,6 +109,11 @@ public class HttpManager {
return gson.toJson(obj);
}
+ /**
+ * doGet
+ *
+ * @return response
+ */
public String doSendGet(String url) {
try {
HttpGet get = getHttpGet(url);
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentMain.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentMain.java
index 0cd1d45ca..0fb9e6cab 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentMain.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentMain.java
@@ -55,7 +55,7 @@ public class AgentMain {
/**
* Print help information
*
- * @param opts - options
+ * @param opts options
*/
private static void help(Options opts) {
HelpFormatter formatter = new HelpFormatter();
@@ -66,7 +66,7 @@ public class AgentMain {
/**
* Init options
*
- * @param args - argument
+ * @param args argument
* @return command line
*/
public static CommandLine initOptions(String[] args) {
@@ -84,7 +84,7 @@ public class AgentMain {
/**
* Init agent conf
*
- * @param cl - commandline
+ * @param cl commandline
*/
public static void initAgentConf(CommandLine cl) {
AgentConfiguration conf = AgentConfiguration.getAgentConf();
@@ -101,7 +101,7 @@ public class AgentMain {
/**
* Stopping agent gracefully if get killed.
*
- * @param manager - agent manager
+ * @param manager agent manager
*/
private static void stopManagerIfKilled(AgentManager manager) {
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
@@ -117,7 +117,7 @@ public class AgentMain {
/**
* Main entrance.
*
- * @param args - arguments
+ * @param args arguments
* @throws Exception exceptions
*/
public static void main(String[] args) throws Exception {
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentManager.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentManager.java
index 64b1cff4e..5708b6a33 100755
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentManager.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/AgentManager.java
@@ -17,13 +17,6 @@
package org.apache.inlong.agent.core;
-import static org.apache.inlong.agent.constant.AgentConstants.AGENT_CONF_PARENT;
-import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_AGENT_CONF_PARENT;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_TRIGGER;
-
-import java.lang.reflect.Constructor;
-import java.util.List;
-
import org.apache.inlong.agent.common.AbstractDaemon;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.conf.JobProfile;
@@ -43,6 +36,13 @@ import org.apache.inlong.agent.db.TriggerProfileDb;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.reflect.Constructor;
+import java.util.List;
+
+import static org.apache.inlong.agent.constant.AgentConstants.AGENT_CONF_PARENT;
+import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_AGENT_CONF_PARENT;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_TRIGGER;
+
/**
* Agent Manager, the bridge for job manager, task manager, db e.t.c it manages agent level operations and communicates
* with outside system.
@@ -89,8 +89,6 @@ public class AgentManager extends AbstractDaemon {
/**
* init fetch by class name
- *
- * @return
*/
private ProfileFetcher initFetcher(AgentManager agentManager) {
try {
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
index c3727eddd..035394e19 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
@@ -45,6 +45,9 @@ import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_HE
import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_REPORTSNAPSHOT_HTTP_PATH;
import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_VIP_HTTP_PREFIX_PATH;
+/**
+ * report heartbeat to inlong-manager
+ */
public class HeartbeatManager extends AbstractDaemon {
private static final Logger LOGGER = LoggerFactory.getLogger(HeartbeatManager.class);
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigJetty.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigJetty.java
index 6e43e2bf2..541acd666 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigJetty.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigJetty.java
@@ -17,10 +17,6 @@
package org.apache.inlong.agent.core.conf;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_SOURCE_TYPE;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_TRIGGER;
-
-import java.io.Closeable;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.conf.TriggerProfile;
@@ -36,6 +32,11 @@ import org.eclipse.jetty.servlet.ServletHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.Closeable;
+
+import static org.apache.inlong.agent.constant.JobConstants.JOB_SOURCE_TYPE;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_TRIGGER;
+
/**
* start http server and get job/agent config via http
*/
@@ -63,8 +64,8 @@ public class ConfigJetty implements Closeable {
private void initJetty() throws Exception {
ServerConnector connector = new ServerConnector(this.server);
connector.setPort(conf.getInt(
- AgentConstants.AGENT_HTTP_PORT, AgentConstants.DEFAULT_AGENT_HTTP_PORT));
- server.setConnectors(new Connector[] { connector });
+ AgentConstants.AGENT_HTTP_PORT, AgentConstants.DEFAULT_AGENT_HTTP_PORT));
+ server.setConnectors(new Connector[]{connector});
ServletHandler servletHandler = new ServletHandler();
ServletHolder holder = new ServletHolder(new ConfigServlet(this));
@@ -73,16 +74,23 @@ public class ConfigJetty implements Closeable {
server.start();
}
+ /**
+ * store job conf:
+ * 1. if it's trigger job, store it in local db;
+ * 2. for other job, submit it to jobManager to be executed
+ *
+ * @param jobProfile JobProfile
+ */
public void storeJobConf(JobProfile jobProfile) {
// store job conf to bdb
if (jobProfile != null) {
// trigger job is a special kind of job
if (jobProfile.hasKey(JOB_TRIGGER)) {
triggerManager.submitTrigger(
- TriggerProfile.parseJsonStr(jobProfile.toJsonStr()));
+ TriggerProfile.parseJsonStr(jobProfile.toJsonStr()));
} else {
TaskTypeEnum taskType = TaskTypeEnum
- .getTaskType(jobProfile.getInt(JOB_SOURCE_TYPE));
+ .getTaskType(jobProfile.getInt(JOB_SOURCE_TYPE));
switch (taskType) {
case SQL:
jobManager.submitSqlJobProfile(jobProfile);
@@ -110,7 +118,8 @@ public class ConfigJetty implements Closeable {
/**
* delete job from conf
- * @param jobProfile
+ *
+ * @param jobProfile JobProfile
*/
public void deleteJobConf(JobProfile jobProfile) {
if (jobProfile != null) {
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigServlet.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigServlet.java
index 0c0c028a2..200a5877b 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigServlet.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/conf/ConfigServlet.java
@@ -18,17 +18,21 @@
package org.apache.inlong.agent.core.conf;
import com.google.gson.Gson;
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.PrintWriter;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
import org.apache.commons.io.IOUtils;
import org.apache.inlong.agent.conf.JobProfile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+/**
+ * Handle config http request
+ */
public class ConfigServlet extends HttpServlet {
private static final Logger LOGGER = LoggerFactory.getLogger(ConfigServlet.class);
@@ -48,8 +52,15 @@ public class ConfigServlet extends HttpServlet {
this.configHandler = configHandler;
}
+ /**
+ * write http response
+ *
+ * @param response HttpServletResponse
+ * @param result ResponseResult
+ * @throws IOException
+ */
public void responseToJson(HttpServletResponse response,
- ResponseResult result) throws IOException {
+ ResponseResult result) throws IOException {
response.setContentType(CONTENT_TYPE);
response.setCharacterEncoding(CHARSET_TYPE);
String jsonStr = gson.toJson(result);
@@ -60,7 +71,8 @@ public class ConfigServlet extends HttpServlet {
/**
* handle path of "/config/job"
- * @param jobProfileStr - job profile string
+ *
+ * @param jobProfileStr job profile string
*/
private void handleJob(String jobProfileStr) {
JobProfile jobProfile = JobProfile.parseJsonStr(jobProfileStr);
@@ -79,8 +91,8 @@ public class ConfigServlet extends HttpServlet {
/**
* handle post requests.
*
- * @param req - request
- * @param resp - response
+ * @param req request
+ * @param resp response
*/
@Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/Job.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/Job.java
index 12cfd7311..42b127636 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/Job.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/Job.java
@@ -17,8 +17,6 @@
package org.apache.inlong.agent.core.job;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.constant.JobConstants;
import org.apache.inlong.agent.core.task.Task;
@@ -29,6 +27,9 @@ import org.apache.inlong.agent.plugin.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* job meta definition, job will be split into several tasks.
*/
@@ -47,7 +48,7 @@ public class Job {
this.jobConf = jobConf;
this.name = jobConf.get(JobConstants.JOB_NAME, JobConstants.DEFAULT_JOB_NAME);
this.description = jobConf.get(
- JobConstants.JOB_DESCRIPTION, JobConstants.DEFAULT_JOB_DESCRIPTION);
+ JobConstants.JOB_DESCRIPTION, JobConstants.DEFAULT_JOB_DESCRIPTION);
this.jobInstanceId = jobConf.get(JobConstants.JOB_INSTANCE_ID);
}
@@ -75,14 +76,19 @@ public class Job {
this.jobInstanceId = jobInstanceId;
}
+ /**
+ * split a job into multi tasks, each task has its own reader, writer and channel
+ *
+ * @return taskList
+ */
public List<Task> createTasks() {
List<Task> taskList = new ArrayList<>();
int index = 0;
try {
LOGGER.info("job id: {}, source: {}, channel: {}, sink: {}",
- getJobInstanceId(), jobConf.get(JobConstants.JOB_SOURCE_CLASS),
- jobConf.get(JobConstants.JOB_CHANNEL),
- jobConf.get(JobConstants.JOB_SINK));
+ getJobInstanceId(), jobConf.get(JobConstants.JOB_SOURCE_CLASS),
+ jobConf.get(JobConstants.JOB_CHANNEL),
+ jobConf.get(JobConstants.JOB_SINK));
Source source = (Source) Class.forName(jobConf.get(JobConstants.JOB_SOURCE_CLASS)).newInstance();
for (Reader reader : source.split(jobConf)) {
Sink writer = (Sink) Class.forName(jobConf.get(JobConstants.JOB_SINK)).newInstance();
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
index 79b9c1609..315594121 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
@@ -46,8 +46,8 @@ import static org.apache.inlong.agent.constant.AgentConstants.JOB_DB_CACHE_TIME;
import static org.apache.inlong.agent.constant.AgentConstants.JOB_NUMBER_LIMIT;
import static org.apache.inlong.agent.constant.JobConstants.JOB_ID;
import static org.apache.inlong.agent.constant.JobConstants.JOB_ID_PREFIX;
-import static org.apache.inlong.agent.constant.JobConstants.SQL_JOB_ID;
import static org.apache.inlong.agent.constant.JobConstants.JOB_INSTANCE_ID;
+import static org.apache.inlong.agent.constant.JobConstants.SQL_JOB_ID;
/**
* JobManager maintains lots of jobs, and communicate between server and task manager.
@@ -55,9 +55,6 @@ import static org.apache.inlong.agent.constant.JobConstants.JOB_INSTANCE_ID;
public class JobManager extends AbstractDaemon {
private static final Logger LOGGER = LoggerFactory.getLogger(JobManager.class);
-
- // key is job instance id.
- private ConcurrentHashMap<String, JobWrapper> jobs;
// jobs which are not accepted by running pool.
private final ConcurrentHashMap<String, Job> pendingJobs;
// job thread pool
@@ -66,17 +63,18 @@ public class JobManager extends AbstractDaemon {
private final int monitorInterval;
private final long jobDbCacheTime;
private final long jobDbCacheCheckInterval;
-
// job profile db is only used to recover instance which is not finished running.
private final JobProfileDb jobProfileDb;
private final JobMetrics jobMetrics;
private final AtomicLong index = new AtomicLong(0);
private final long jobMaxSize;
+ // key is job instance id.
+ private ConcurrentHashMap<String, JobWrapper> jobs;
/**
* init job manager
*
- * @param agentManager - agent manager
+ * @param agentManager agent manager
*/
public JobManager(AgentManager agentManager, JobProfileDb jobProfileDb) {
this.jobProfileDb = jobProfileDb;
@@ -107,7 +105,7 @@ public class JobManager extends AbstractDaemon {
/**
* submit job to work thread.
*
- * @param job - job
+ * @param job job
*/
private void addJob(Job job) {
try {
@@ -129,7 +127,7 @@ public class JobManager extends AbstractDaemon {
/**
* add file job profile
*
- * @param profile - job profile.
+ * @param profile job profile.
*/
public boolean submitFileJobProfile(JobProfile profile) {
return submitJobProfile(profile, false);
@@ -138,7 +136,7 @@ public class JobManager extends AbstractDaemon {
/**
* add file job profile
*
- * @param profile - job profile.
+ * @param profile job profile.
*/
public boolean submitJobProfile(JobProfile profile, boolean singleJob) {
if (!isJobValid(profile)) {
@@ -159,7 +157,7 @@ public class JobManager extends AbstractDaemon {
/**
* add sql job profile
*
- * @param profile - job profile.
+ * @param profile job profile.
*/
public boolean submitSqlJobProfile(JobProfile profile) {
if (isJobValid(profile)) {
@@ -185,6 +183,9 @@ public class JobManager extends AbstractDaemon {
return true;
}
+ /**
+ * whether job size exceeds maxSize
+ */
public boolean isJobOverLimit() {
return jobs.size() >= jobMaxSize;
}
@@ -217,6 +218,9 @@ public class JobManager extends AbstractDaemon {
}
}
+ /**
+ * check pending jobs and submit them
+ */
public Runnable jobStateCheckThread() {
return () -> {
while (isRunnable()) {
@@ -238,8 +242,6 @@ public class JobManager extends AbstractDaemon {
/**
* check local db and delete old tasks.
- *
- * @return
*/
public Runnable dbStorageCheckThread() {
return () -> {
@@ -261,7 +263,7 @@ public class JobManager extends AbstractDaemon {
/**
* mark job as success by job id.
*
- * @param jobId - job id
+ * @param jobId job id
*/
public void markJobAsSuccess(String jobId) {
JobWrapper wrapper = jobs.remove(jobId);
@@ -273,6 +275,11 @@ public class JobManager extends AbstractDaemon {
}
}
+ /**
+ * remove job from jobs, and mark it as failed
+ *
+ * @param jobId job id
+ */
public void markJobAsFailed(String jobId) {
JobWrapper wrapper = jobs.remove(jobId);
if (wrapper != null) {
@@ -290,8 +297,6 @@ public class JobManager extends AbstractDaemon {
/**
* check job existence using job file name
- *
- * @return
*/
public boolean checkJobExsit(String fileName) {
return jobProfileDb.getJobByFileName(fileName) != null;
@@ -299,8 +304,6 @@ public class JobManager extends AbstractDaemon {
/**
* get sql job existence
- *
- * @return
*/
public boolean sqlJobExsit() {
return jobProfileDb.getJobById(SQL_JOB_ID) != null;
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobWrapper.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobWrapper.java
index b8111ad1e..d212a8d20 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobWrapper.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobWrapper.java
@@ -17,13 +17,6 @@
package org.apache.inlong.agent.core.job;
-import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_JOB_VERSION;
-import static org.apache.inlong.agent.constant.AgentConstants.JOB_VERSION;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_OFFSET_DELIMITER;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.constant.AgentConstants;
import org.apache.inlong.agent.core.AgentManager;
@@ -37,6 +30,14 @@ import org.apache.inlong.common.db.CommandEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_JOB_VERSION;
+import static org.apache.inlong.agent.constant.AgentConstants.JOB_VERSION;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_OFFSET_DELIMITER;
+
/**
* JobWrapper is used in JobManager, it defines the life cycle of
* running job and maintains the state of job.
@@ -49,9 +50,8 @@ public class JobWrapper extends AbstractStateWrapper {
private final TaskManager taskManager;
private final JobManager jobManager;
private final Job job;
- private CommandDb db;
-
private final List<Task> allTasks;
+ private CommandDb db;
public JobWrapper(AgentManager manager, Job job) {
super();
@@ -111,7 +111,7 @@ public class JobWrapper extends AbstractStateWrapper {
/**
* get snapshot of each task
*
- * @return
+ * @return snapshot string
*/
public String getSnapshot() {
List<String> snapshotList = new ArrayList<>();
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
index 17258b782..39deb0c2f 100755
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
@@ -17,12 +17,6 @@
package org.apache.inlong.agent.core.task;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import org.apache.inlong.agent.common.AbstractDaemon;
import org.apache.inlong.agent.common.AgentThreadFactory;
import org.apache.inlong.agent.conf.AgentConfiguration;
@@ -33,6 +27,13 @@ import org.apache.inlong.agent.utils.ConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
/**
* Task manager maintains lots of tasks and communicate with job level components.
* It also provide functions to execute commands from job level like killing/submit tasks.
@@ -55,15 +56,15 @@ public class TaskManager extends AbstractDaemon {
/**
* Init task manager.
*
- * @param agentManager - agent manager
+ * @param agentManager agent manager
*/
public TaskManager(AgentManager agentManager) {
this.agentManager = agentManager;
this.runningPool = new ThreadPoolExecutor(
- 0, Integer.MAX_VALUE,
- 60L, TimeUnit.SECONDS,
- new SynchronousQueue<>(),
- new AgentThreadFactory("task"));
+ 0, Integer.MAX_VALUE,
+ 60L, TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new AgentThreadFactory("task"));
// metric for task level
if (ConfigUtil.isPrometheusEnabled()) {
this.taskMetrics = new TaskPrometheusMetrics();
@@ -75,16 +76,16 @@ public class TaskManager extends AbstractDaemon {
AgentConfiguration conf = AgentConfiguration.getAgentConf();
retryTasks = new LinkedBlockingQueue<>(
conf.getInt(
- AgentConstants.TASK_RETRY_MAX_CAPACITY, AgentConstants.DEFAULT_TASK_RETRY_MAX_CAPACITY));
+ AgentConstants.TASK_RETRY_MAX_CAPACITY, AgentConstants.DEFAULT_TASK_RETRY_MAX_CAPACITY));
monitorInterval = conf.getInt(
- AgentConstants.TASK_MONITOR_INTERVAL, AgentConstants.DEFAULT_TASK_MONITOR_INTERVAL);
+ AgentConstants.TASK_MONITOR_INTERVAL, AgentConstants.DEFAULT_TASK_MONITOR_INTERVAL);
taskRetryMaxTime = conf
.getInt(AgentConstants.TASK_RETRY_SUBMIT_WAIT_SECONDS,
- AgentConstants.DEFAULT_TASK_RETRY_SUBMIT_WAIT_SECONDS);
+ AgentConstants.DEFAULT_TASK_RETRY_SUBMIT_WAIT_SECONDS);
taskMaxCapacity = conf.getInt(
- AgentConstants.TASK_RETRY_MAX_CAPACITY, AgentConstants.DEFAULT_TASK_RETRY_MAX_CAPACITY);
+ AgentConstants.TASK_RETRY_MAX_CAPACITY, AgentConstants.DEFAULT_TASK_RETRY_MAX_CAPACITY);
waitTime = conf.getLong(
- AgentConstants.THREAD_POOL_AWAIT_TIME, AgentConstants.DEFAULT_THREAD_POOL_AWAIT_TIME);
+ AgentConstants.THREAD_POOL_AWAIT_TIME, AgentConstants.DEFAULT_THREAD_POOL_AWAIT_TIME);
}
/**
@@ -103,7 +104,7 @@ public class TaskManager extends AbstractDaemon {
/**
* submit task, wait if task queue is full.
*
- * @param task - task
+ * @param task task
*/
public void submitTask(Task task) {
TaskWrapper taskWrapper = new TaskWrapper(agentManager, task);
@@ -134,7 +135,7 @@ public class TaskManager extends AbstractDaemon {
/**
* retry task.
*
- * @param wrapper - task wrapper
+ * @param wrapper task wrapper
*/
private boolean addRetryTask(TaskWrapper wrapper) {
LOGGER.info("retry submit task {}", wrapper.getTask().getTaskId());
@@ -156,8 +157,8 @@ public class TaskManager extends AbstractDaemon {
/**
* Check whether task is finished
*
- * @param taskId - task id
- * @return - true if task is finished otherwise false
+ * @param taskId task id
+ * @return true if task is finished otherwise false
*/
public boolean isTaskFinished(String taskId) {
TaskWrapper wrapper = tasks.get(taskId);
@@ -184,7 +185,7 @@ public class TaskManager extends AbstractDaemon {
/**
* Remove task and wait task to finish by task id
*
- * @param taskId - task id
+ * @param taskId task id
*/
public void removeTask(String taskId) {
taskMetrics.decRunningTaskCount();
@@ -198,7 +199,6 @@ public class TaskManager extends AbstractDaemon {
* kill task
*
* @param task task
- * @return
*/
public boolean killTask(Task task) {
// kill running tasks.
@@ -213,7 +213,7 @@ public class TaskManager extends AbstractDaemon {
/**
* Thread for checking whether task should retry.
*
- * @return - runnable thread
+ * @return runnable thread
*/
public Runnable createTaskMonitorThread() {
return () -> {
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskPositionManager.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskPositionManager.java
index 1ee670e23..b3ca813bf 100644
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskPositionManager.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskPositionManager.java
@@ -40,12 +40,11 @@ import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_FE
public class TaskPositionManager extends AbstractDaemon {
private static final Logger LOGGER = LoggerFactory.getLogger(TaskPositionManager.class);
-
+ private static volatile TaskPositionManager taskPositionManager = null;
private final AgentManager agentManager;
private final JobProfileDb jobConfDb;
- private ConcurrentHashMap<String, ConcurrentHashMap<String, Long>> jobTaskPositionMap;
private final AgentConfiguration conf;
- private static volatile TaskPositionManager taskPositionManager = null;
+ private ConcurrentHashMap<String, ConcurrentHashMap<String, Long>> jobTaskPositionMap;
private TaskPositionManager(AgentManager agentManager) {
this.conf = AgentConfiguration.getAgentConf();
@@ -56,9 +55,6 @@ public class TaskPositionManager extends AbstractDaemon {
/**
* task position manager singleton, can only generated by agent manager
- *
- * @param agentManager
- * @return
*/
public static TaskPositionManager getTaskPositionManager(AgentManager agentManager) {
if (taskPositionManager == null) {
@@ -73,8 +69,6 @@ public class TaskPositionManager extends AbstractDaemon {
/**
* get taskPositionManager singleton
- *
- * @return
*/
public static TaskPositionManager getTaskPositionManager() {
if (taskPositionManager == null) {
@@ -134,6 +128,11 @@ public class TaskPositionManager extends AbstractDaemon {
waitForTerminate();
}
+ /**
+ * update job sink position
+ *
+ * @param size add this size to beforePosition
+ */
public void updateSinkPosition(String jobInstanceId, String sourcePath, long size) {
ConcurrentHashMap<String, Long> positionTemp = new ConcurrentHashMap<>();
ConcurrentHashMap<String, Long> position = jobTaskPositionMap.putIfAbsent(jobInstanceId, positionTemp);
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
index e0555433c..d21ae68e8 100755
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
@@ -17,16 +17,6 @@
package org.apache.inlong.agent.core.task;
-import static org.apache.inlong.agent.constant.JobConstants.DEFAULT_JOB_READ_WAIT_TIMEOUT;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_READ_WAIT_TIMEOUT;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.inlong.agent.common.AgentThreadFactory;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.constant.AgentConstants;
@@ -39,16 +29,25 @@ import org.apache.inlong.agent.utils.AgentUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.inlong.agent.constant.JobConstants.DEFAULT_JOB_READ_WAIT_TIMEOUT;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_READ_WAIT_TIMEOUT;
+
/**
* TaskWrapper is used in taskManager, it maintains the life cycle of
* running task.
*/
public class TaskWrapper extends AbstractStateWrapper {
- private static final Logger LOGGER = LoggerFactory.getLogger(TaskWrapper.class);
public static final int WAIT_FINISH_TIME_OUT = 1;
public static final int WAIT_BEGIN_TIME_MINUTE = 1;
-
+ private static final Logger LOGGER = LoggerFactory.getLogger(TaskWrapper.class);
private final TaskManager taskManager;
private final Task task;
@@ -101,7 +100,7 @@ public class TaskWrapper extends AbstractStateWrapper {
AgentUtils.silenceSleepInMs(readWaitTime);
}
LOGGER.info("read end, task exception status is {}, read finish status is {}", isException(),
- task.isReadFinished());
+ task.isReadFinished());
// write end message
task.getChannel().push(new EndMessage());
task.getReader().destroy();
@@ -157,7 +156,6 @@ public class TaskWrapper extends AbstractStateWrapper {
}
-
/**
* destroy task
*/
@@ -169,7 +167,7 @@ public class TaskWrapper extends AbstractStateWrapper {
/**
* whether task retry times exceed max retry time.
*
- * @return - whether should retry
+ * @return whether should retry
*/
boolean shouldRetry() {
return retryTime.get() < maxRetryTime;
diff --git a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/trigger/TriggerManager.java b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/trigger/TriggerManager.java
index 4ba183c9a..1e05d58dd 100755
--- a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/trigger/TriggerManager.java
+++ b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/trigger/TriggerManager.java
@@ -17,15 +17,6 @@
package org.apache.inlong.agent.core.trigger;
-import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_TRIGGER_MAX_RUNNING_NUM;
-import static org.apache.inlong.agent.constant.AgentConstants.TRIGGER_MAX_RUNNING_NUM;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_ID;
-import static org.apache.inlong.agent.constant.JobConstants.TRIGGER_ONLY_ONE_JOB;
-
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
import org.apache.inlong.agent.common.AbstractDaemon;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.conf.JobProfile;
@@ -39,14 +30,23 @@ import org.apache.inlong.agent.plugin.Trigger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_TRIGGER_MAX_RUNNING_NUM;
+import static org.apache.inlong.agent.constant.AgentConstants.TRIGGER_MAX_RUNNING_NUM;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_ID;
+import static org.apache.inlong.agent.constant.JobConstants.TRIGGER_ONLY_ONE_JOB;
+
/**
* manager for triggers.
*/
public class TriggerManager extends AbstractDaemon {
- private static final Logger LOGGER = LoggerFactory.getLogger(TriggerManager.class);
public static final int JOB_CHECK_INTERVAL = 1;
-
+ private static final Logger LOGGER = LoggerFactory.getLogger(TriggerManager.class);
private final AgentManager manager;
private final TriggerProfileDb triggerProfileDB;
private final ConcurrentHashMap<String, Trigger> triggerMap;
@@ -68,7 +68,8 @@ public class TriggerManager extends AbstractDaemon {
/**
* submit trigger profile.
- * @param triggerProfile - trigger profile
+ *
+ * @param triggerProfile trigger profile
*/
public boolean addTrigger(TriggerProfile triggerProfile) {
try {
@@ -93,11 +94,16 @@ public class TriggerManager extends AbstractDaemon {
return triggerMap.get(triggerId);
}
+ /**
+ * check trigger config, store it to db, and submit this trigger to be executed
+ *
+ * @return true if success
+ */
public boolean submitTrigger(TriggerProfile triggerProfile) {
// make sure all required key exists.
if (!triggerProfile.allRequiredKeyExist() || this.triggerMap.size() > maxRunningNum) {
LOGGER.error("trigger {} not all required key exists or size {} exceed {}",
- triggerProfile.toJsonStr(), this.triggerMap.size(), maxRunningNum);
+ triggerProfile.toJsonStr(), this.triggerMap.size(), maxRunningNum);
return false;
}
triggerProfileDB.storeTrigger(triggerProfile);
@@ -131,12 +137,11 @@ public class TriggerManager extends AbstractDaemon {
/**
* delete jobs generated by the trigger
- * @param triggerId
*/
private void deleteRelatedJobs(String triggerId) {
LOGGER.info("start to delete related jobs in triggerId {}", triggerId);
ConcurrentHashMap<String, JobProfile> jobProfiles =
- triggerJobMap.get(triggerId);
+ triggerJobMap.get(triggerId);
if (jobProfiles != null) {
LOGGER.info("trigger can only run one job, stop the others {}", jobProfiles.keySet());
jobProfiles.keySet().forEach(this::deleteJob);
@@ -155,7 +160,7 @@ public class TriggerManager extends AbstractDaemon {
triggerJobMap.forEach((s, jobProfiles) -> {
for (String jobId : jobProfiles.keySet()) {
Map<String, JobWrapper> jobs =
- manager.getJobManager().getJobs();
+ manager.getJobManager().getJobs();
if (jobs.get(jobId) == null) {
triggerJobMap.remove(jobId);
}
@@ -172,14 +177,12 @@ public class TriggerManager extends AbstractDaemon {
/**
* need to put profile in triggerJobMap
- * @param triggerId
- * @param profile
*/
private void addToTriggerMap(String triggerId, JobProfile profile) {
ConcurrentHashMap<String, JobProfile> tmpList =
- new ConcurrentHashMap<>();
+ new ConcurrentHashMap<>();
ConcurrentHashMap<String, JobProfile> jobWrappers =
- triggerJobMap.putIfAbsent(triggerId, tmpList);
+ triggerJobMap.putIfAbsent(triggerId, tmpList);
if (jobWrappers == null) {
jobWrappers = tmpList;
}
@@ -188,7 +191,8 @@ public class TriggerManager extends AbstractDaemon {
/**
* delete trigger by trigger profile.
- * @param triggerId - trigger profile.
+ *
+ * @param triggerId trigger profile.
*/
public boolean deleteTrigger(String triggerId) {
LOGGER.info("delete trigger {}", triggerId);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
index d54a4bc9e..56769a30f 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
@@ -17,10 +17,6 @@
package org.apache.inlong.agent.plugin.channel;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.constant.AgentConstants;
import org.apache.inlong.agent.plugin.Channel;
@@ -33,30 +29,32 @@ import org.apache.inlong.agent.utils.ConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * memory channel
+ */
public class MemoryChannel implements Channel {
private static final Logger LOGGER = LoggerFactory.getLogger(MemoryChannel.class);
private static final String MEMORY_CHANNEL_TAG_NAME = "AgentMemoryPlugin";
-
- private LinkedBlockingQueue<Message> queue;
-
- private final PluginMetric pluginMetricNew;
private static AtomicLong metricsIndex = new AtomicLong(0);
+ private final PluginMetric pluginMetricNew;
+ private LinkedBlockingQueue<Message> queue;
public MemoryChannel() {
if (ConfigUtil.isPrometheusEnabled()) {
this.pluginMetricNew = new PluginPrometheusMetric(AgentUtils.getUniqId(
- MEMORY_CHANNEL_TAG_NAME, metricsIndex.incrementAndGet()));
+ MEMORY_CHANNEL_TAG_NAME, metricsIndex.incrementAndGet()));
} else {
this.pluginMetricNew = new PluginJmxMetric(AgentUtils.getUniqId(
- MEMORY_CHANNEL_TAG_NAME, metricsIndex.incrementAndGet()));
+ MEMORY_CHANNEL_TAG_NAME, metricsIndex.incrementAndGet()));
}
}
- /**
- * {@inheritDoc}
- */
@Override
public void push(Message message) {
try {
@@ -91,9 +89,6 @@ public class MemoryChannel implements Channel {
return false;
}
- /**
- * {@inheritDoc}
- */
@Override
public Message pull(long timeout, TimeUnit unit) {
try {
@@ -113,7 +108,7 @@ public class MemoryChannel implements Channel {
public void init(JobProfile jobConf) {
queue = new LinkedBlockingQueue<>(
jobConf.getInt(AgentConstants.CHANNEL_MEMORY_CAPACITY,
- AgentConstants.DEFAULT_CHANNEL_MEMORY_CAPACITY));
+ AgentConstants.DEFAULT_CHANNEL_MEMORY_CAPACITY));
}
@Override
@@ -122,9 +117,9 @@ public class MemoryChannel implements Channel {
queue.clear();
}
LOGGER.info("destroy channel, memory channel metric, readNum: {}, readSuccessNum: {}, "
- + "readFailedNum: {}, sendSuccessNum: {}, sendFailedNum: {}",
- pluginMetricNew.getReadNum(), pluginMetricNew.getReadSuccessNum(),
- pluginMetricNew.getReadFailedNum(), pluginMetricNew.getSendSuccessNum(),
- pluginMetricNew.getSendFailedNum());
+ + "readFailedNum: {}, sendSuccessNum: {}, sendFailedNum: {}",
+ pluginMetricNew.getReadNum(), pluginMetricNew.getReadSuccessNum(),
+ pluginMetricNew.getReadFailedNum(), pluginMetricNew.getSendSuccessNum(),
+ pluginMetricNew.getSendFailedNum());
}
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
index 27fb56495..b6a06e56b 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
@@ -209,6 +209,9 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
return null;
}
+ /**
+ * request manager to get manager vipUrl list, and store it to local file
+ */
public void requestTdmList() {
JsonObject result = getResultData(httpManager.doSendGet(managerVipUrl));
JsonArray data = result.get(AGENT_MANAGER_RETURN_PARAM_DATA).getAsJsonArray();
@@ -377,7 +380,7 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
Collection<File> suitFiles = PluginUtils.findSuitFiles(triggerProfile);
// filter files exited before
List<File> pendingFiles = suitFiles.stream().filter(file ->
- !agentManager.getJobManager().checkJobExsit(file.getAbsolutePath()))
+ !agentManager.getJobManager().checkJobExsit(file.getAbsolutePath()))
.collect(Collectors.toList());
for (File pendingFile : pendingFiles) {
JobProfile copiedProfile = copyJobProfile(triggerProfile, dataTime,
@@ -444,8 +447,8 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
/**
* fetch manager list, make sure it's not throwing exceptions
*
- * @param isInitial - is initial
- * @param retryTime - retry time
+ * @param isInitial is initial
+ * @param retryTime retry time
*/
private void fetchTdmList(boolean isInitial, int retryTime) {
if (retryTime > MAX_RETRY) {
@@ -496,7 +499,7 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
/**
* request manager to get trigger profiles.
*
- * @return - trigger profile list
+ * @return trigger profile list
*/
@Override
public List<TriggerProfile> getTriggerProfiles() {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
index acb03ae98..3b49dfb9c 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
@@ -31,12 +31,10 @@ import java.util.List;
*/
public class ManagerResultFormatter {
+ public static final String SUCCESS_CODE = "true";
private static final Logger LOGGER = LoggerFactory.getLogger(ManagerResultFormatter.class);
-
private static final String RESULT_CODE = "success";
private static final String RESULT_DATA = "data";
- public static final String SUCCESS_CODE = "true";
-
private static final Gson GSON = new Gson();
/**
@@ -58,8 +56,8 @@ public class ManagerResultFormatter {
/**
* get random list of base list.
*
- * @param baseList - base list
- * @param num - max Num
+ * @param baseList base list
+ * @param num max Num
* @return random list
*/
public static <T> List<T> getRandomList(List<T> baseList, int num) {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/filter/DateFormatRegex.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/filter/DateFormatRegex.java
index da9604e5c..ceaaacf73 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/filter/DateFormatRegex.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/filter/DateFormatRegex.java
@@ -17,19 +17,21 @@
package org.apache.inlong.agent.plugin.filter;
-import java.io.File;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.agent.plugin.Filter;
import org.apache.inlong.agent.utils.AgentUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* date format with regex string for absolute file path.
*/
public class DateFormatRegex implements Filter {
+
private static final Logger LOGGER = LoggerFactory.getLogger(DateFormatRegex.class);
private static final String YEAR = "YYYY";
@@ -53,8 +55,6 @@ public class DateFormatRegex implements Filter {
/**
* set regex with current time
- * @param regex
- * @return
*/
public static DateFormatRegex ofRegex(String regex) {
DateFormatRegex dateFormatRegex = new DateFormatRegex();
@@ -68,6 +68,9 @@ public class DateFormatRegex implements Filter {
return AgentUtils.regexMatch(file.getAbsolutePath(), formattedRegex);
}
+ /**
+ * set timeOffset
+ */
public DateFormatRegex withOffset(String timeOffset) {
String number = StringUtils.substring(timeOffset, 0, timeOffset.length() - 1);
String mark = StringUtils.substring(timeOffset, timeOffset.length() - 1);
@@ -95,19 +98,17 @@ public class DateFormatRegex implements Filter {
/**
* set regex with given time, replace the YYYYDDMM pattern with given time
- * @param regex
- * @param time
*/
public void setRegexWithTime(String regex, String time) {
String[] regexList = StringUtils.splitByWholeSeparatorPreserveAllTokens(regex,
- File.separator, 0);
+ File.separator, 0);
List<String> formattedList = new ArrayList<>();
for (String regexStr : regexList) {
if (regexStr.contains(YEAR)) {
String tmpRegexStr = regexStr.replace(YEAR, time.substring(0, 4))
- .replace(MONTH, time.substring(4, 6))
- .replace(DAY, time.substring(6, 8))
- .replace(HOUR, time.substring(8, 10));
+ .replace(MONTH, time.substring(4, 6))
+ .replace(DAY, time.substring(6, 8))
+ .replace(HOUR, time.substring(8, 10));
formattedList.add(tmpRegexStr);
formattedTime = time;
} else {
@@ -118,9 +119,12 @@ public class DateFormatRegex implements Filter {
LOGGER.info("updated formatted regex is {}", this.formattedRegex);
}
+ /**
+ * set regex with current time, replace the YYYYDDMM pattern with current time
+ */
public void setRegexWithCurrentTime(String regex) {
String currentTime = AgentUtils.formatCurrentTimeWithOffset(NORMAL_FORMATTER,
- dayOffset, hourOffset, minuteOffset);
+ dayOffset, hourOffset, minuteOffset);
setRegexWithTime(regex, currentTime);
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/PackProxyMessage.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/PackProxyMessage.java
index 8c0944f31..fabcaead6 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/PackProxyMessage.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/PackProxyMessage.java
@@ -17,16 +17,17 @@
package org.apache.inlong.agent.plugin.message;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.inlong.agent.message.ProxyMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.inlong.agent.message.ProxyMessage;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Handle List of BusMessage, which belong to the same stream id.
@@ -36,32 +37,26 @@ public class PackProxyMessage {
private static final Logger LOGGER = LoggerFactory.getLogger(PackProxyMessage.class);
private final String streamId;
-
- private int currentSize;
-
private final int maxPackSize;
-
private final int maxQueueNumber;
// ms
private final int cacheTimeout;
-
+ // streamId -> list of proxyMessage
+ private final LinkedBlockingQueue<ProxyMessage> messageQueue;
+ private final AtomicLong queueSize = new AtomicLong(0);
+ private int currentSize;
/**
* extra map used when sending to dataproxy
*/
private Map<String, String> extraMap = new HashMap<>();
-
- // streamId -> list of proxyMessage
- private final LinkedBlockingQueue<ProxyMessage> messageQueue;
-
private volatile long currentCacheTime = System.currentTimeMillis();
- private final AtomicLong queueSize = new AtomicLong(0);
/**
* Init PackBusMessage
*
- * @param maxPackSize - max pack size for one inlongGroupId
- * @param cacheTimeout - cache timeout for one proxy message
- * @param streamId - streamId
+ * @param maxPackSize max pack size for one inlongGroupId
+ * @param cacheTimeout cache timeout for one proxy message
+ * @param streamId streamId
*/
public PackProxyMessage(int maxPackSize, int maxQueueNumber, int cacheTimeout, String streamId) {
this.maxPackSize = maxPackSize;
@@ -94,7 +89,7 @@ public class PackProxyMessage {
try {
if (queueIsFull()) {
LOGGER.warn("message queue is greater than {}, stop adding message, "
- + "maybe proxy get stuck", maxQueueNumber);
+ + "maybe proxy get stuck", maxQueueNumber);
}
messageQueue.put(message);
queueSize.addAndGet(message.getBody().length);
@@ -105,7 +100,6 @@ public class PackProxyMessage {
/**
* check message queue is empty or not
- * @return
*/
public boolean isEmpty() {
return messageQueue.isEmpty();
@@ -114,8 +108,7 @@ public class PackProxyMessage {
/**
* Fetch batch of proxy message, timeout message or max number of list satisfied.
*
- * @return map of message list, key is stream id for the batch.
- * return null if there are no valid messages.
+ * @return map of message list, key is stream id for the batch; return null if there are no valid messages.
*/
public Pair<String, List<byte[]>> fetchBatch() {
// if queue is nearly full or package size is satisfied or timeout
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/SequentialID.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/SequentialID.java
index 8f2e44943..879e42e57 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/SequentialID.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/SequentialID.java
@@ -17,10 +17,11 @@
package org.apache.inlong.agent.plugin.message;
+import org.apache.inlong.agent.utils.AgentUtils;
+
import java.net.InetAddress;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
-import org.apache.inlong.agent.utils.AgentUtils;
/**
* Generate uniq sequential id, reset base id if max number
@@ -28,14 +29,13 @@ import org.apache.inlong.agent.utils.AgentUtils;
*/
public class SequentialID {
+ private static final int MAX_ID = 2000000000;
+ private static final String IP_HEX = getHex();
+ private static SequentialID uniqueSequentialID = null;
private final AtomicInteger id = new AtomicInteger(1);
private final ReentrantLock idLock = new ReentrantLock();
private final AtomicInteger uid = new AtomicInteger(1);
private final ReentrantLock uidLock = new ReentrantLock();
- private static final int MAX_ID = 2000000000;
-
- private static SequentialID uniqueSequentialID = null;
- private static final String IP_HEX = getHex();
private SequentialID() {
@@ -57,6 +57,9 @@ public class SequentialID {
return "00000000";
}
+ /**
+ * get SequentialID single instance
+ */
public static synchronized SequentialID getInstance() {
if (uniqueSequentialID == null) {
@@ -77,6 +80,9 @@ public class SequentialID {
}
}
+ /**
+ * get next uuid
+ */
public String getNextUuid() {
uidLock.lock();
try {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java
index 7aeef9cf6..1d1035a2f 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java
@@ -17,14 +17,7 @@
package org.apache.inlong.agent.plugin.sinks;
-import static org.apache.inlong.agent.constant.AgentConstants.AGENT_MESSAGE_FILTER_CLASSNAME;
-import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_GROUP_ID;
-import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_STREAM_ID;
-import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
-import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
-
import com.google.common.base.Joiner;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.Sink;
@@ -35,6 +28,17 @@ import org.apache.inlong.agent.utils.ConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.inlong.agent.constant.AgentConstants.AGENT_MESSAGE_FILTER_CLASSNAME;
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_STREAM_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+
+/**
+ * abstract sink: sink data to remote data center
+ */
public abstract class AbstractSink implements Sink {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractSink.class);
@@ -42,19 +46,16 @@ public abstract class AbstractSink implements Sink {
protected static SinkMetrics sinkMetric;
protected static SinkMetrics streamMetric;
-
+ private static AtomicLong index = new AtomicLong(0);
protected String inlongGroupId;
-
protected String inlongStreamId;
- private static AtomicLong index = new AtomicLong(0);
-
@Override
public MessageFilter initMessageFilter(JobProfile jobConf) {
if (jobConf.hasKey(AGENT_MESSAGE_FILTER_CLASSNAME)) {
try {
return (MessageFilter) Class.forName(jobConf.get(AGENT_MESSAGE_FILTER_CLASSNAME))
- .getDeclaredConstructor().newInstance();
+ .getDeclaredConstructor().newInstance();
} catch (Exception e) {
LOGGER.error("init message filter error", e);
}
@@ -68,6 +69,11 @@ public abstract class AbstractSink implements Sink {
inlongStreamId = jobConf.get(PROXY_INLONG_STREAM_ID, DEFAULT_PROXY_INLONG_STREAM_ID);
}
+ /**
+ * init sinkMetric
+ *
+ * @param tagName metric tagName
+ */
protected void intMetric(String tagName) {
String label = Joiner.on(",").join(tagName, index.getAndIncrement());
if (ConfigUtil.isPrometheusEnabled()) {
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
index 8d31a380f..235f9730e 100755
--- 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
@@ -52,10 +52,17 @@ import static org.apache.inlong.agent.constant.JobConstants.PROXY_INLONG_STREAM_
import static org.apache.inlong.agent.constant.JobConstants.PROXY_PACKAGE_MAX_SIZE;
import static org.apache.inlong.agent.constant.JobConstants.PROXY_PACKAGE_MAX_TIMEOUT_MS;
+/**
+ * sink message data to inlong-dataproxy
+ */
public class ProxySink extends AbstractSink {
private static final Logger LOGGER = LoggerFactory.getLogger(ProxySink.class);
private static final String PROXY_SINK_TAG_NAME = "AgentProxySinkMetric";
+ private static AtomicLong index = new AtomicLong(0);
+ private final ExecutorService executorService = new ThreadPoolExecutor(1, 1,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>(), new AgentThreadFactory("ProxySink"));
private MessageFilter messageFilter;
private SenderManager senderManager;
private byte[] fieldSplitter;
@@ -66,11 +73,7 @@ public class ProxySink extends AbstractSink {
private int batchFlushInterval;
private int maxQueueNumber;
private boolean syncSend;
- private final ExecutorService executorService = new ThreadPoolExecutor(1, 1,
- 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<>(), new AgentThreadFactory("ProxySink"));
private volatile boolean shutdown = false;
- private static AtomicLong index = new AtomicLong(0);
// key is stream id, value is a batch of messages belong to the same stream id
private ConcurrentHashMap<String, PackProxyMessage> cache;
@@ -89,23 +92,23 @@ public class ProxySink extends AbstractSink {
ProxyMessage proxyMessage = ProxyMessage.parse(message);
// add proxy message to cache.
cache.compute(proxyMessage.getBatchKey(),
- (s, packProxyMessage) -> {
- if (packProxyMessage == null) {
- packProxyMessage = new PackProxyMessage(
- maxBatchSize, maxQueueNumber,
- maxBatchTimeoutMs,
- proxyMessage.getInlongStreamId()
- );
- packProxyMessage.generateExtraMap(syncSend,
- proxyMessage.getDataKey());
- }
- // add message to package proxy
- packProxyMessage.addProxyMessage(proxyMessage);
- //
- return packProxyMessage;
- });
+ (s, packProxyMessage) -> {
+ if (packProxyMessage == null) {
+ packProxyMessage = new PackProxyMessage(
+ maxBatchSize, maxQueueNumber,
+ maxBatchTimeoutMs,
+ proxyMessage.getInlongStreamId()
+ );
+ packProxyMessage.generateExtraMap(syncSend,
+ proxyMessage.getDataKey());
+ }
+ // add message to package proxy
+ packProxyMessage.addProxyMessage(proxyMessage);
+ //
+ return packProxyMessage;
+ });
AuditUtils.add(AuditUtils.AUDIT_ID_AGENT_SEND_SUCCESS,
- inlongGroupId, inlongStreamId, System.currentTimeMillis());
+ inlongGroupId, inlongStreamId, System.currentTimeMillis());
// increment the count of successful sinks
sinkMetric.incSinkSuccessCount();
streamMetric.incSinkSuccessCount();
@@ -122,9 +125,6 @@ public class ProxySink extends AbstractSink {
/**
* extract stream id from message if message filter is presented
- * or use the default stream id
- *
- * @param message
*/
private void extractStreamFromMessage(Message message, byte[] fieldSplitter) {
if (messageFilter != null) {
@@ -143,7 +143,7 @@ public class ProxySink extends AbstractSink {
/**
* flush cache by batch
*
- * @return - thread runner
+ * @return thread runner
*/
private Runnable flushCache() {
return () -> {
@@ -156,14 +156,14 @@ public class ProxySink extends AbstractSink {
long sendTime = AgentUtils.getCurrentTime();
if (syncSend) {
senderManager.sendBatchSync(inlongGroupId, result.getKey(), result.getValue(),
- 0, sendTime, packProxyMessage.getExtraMap());
+ 0, sendTime, packProxyMessage.getExtraMap());
} else {
senderManager.sendBatchAsync(jobInstanceId, inlongGroupId, result.getKey(),
- result.getValue(), 0, sendTime);
+ result.getValue(), 0, sendTime);
}
LOGGER.info("send group id {}, message key {},with message size {}, the job id is {}, "
- + "read source is {} sendTime is {} syncSend {}", inlongGroupId, batchKey,
- result.getRight().size(), jobInstanceId, sourceName, sendTime, syncSend);
+ + "read source is {} sendTime is {} syncSend {}", inlongGroupId, batchKey,
+ result.getRight().size(), jobInstanceId, sourceName, sendTime, syncSend);
}
});
@@ -215,8 +215,6 @@ public class ProxySink extends AbstractSink {
/**
* check whether all stream id messages finished
- *
- * @return
*/
private boolean sinkFinish() {
return cache.values().stream().allMatch(PackProxyMessage::isEmpty);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
index 9f4c467cb..c0183afb6 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
@@ -17,18 +17,7 @@
package org.apache.inlong.agent.plugin.sinks;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_HOST;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_PORT;
-
import io.netty.util.concurrent.DefaultThreadFactory;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.constant.CommonConstants;
@@ -39,13 +28,25 @@ import org.apache.inlong.agent.plugin.metrics.PluginMetric;
import org.apache.inlong.agent.plugin.metrics.PluginPrometheusMetric;
import org.apache.inlong.agent.utils.AgentUtils;
import org.apache.inlong.agent.utils.ConfigUtil;
-import org.apache.inlong.sdk.dataproxy.ProxyClientConfig;
import org.apache.inlong.sdk.dataproxy.DefaultMessageSender;
+import org.apache.inlong.sdk.dataproxy.ProxyClientConfig;
import org.apache.inlong.sdk.dataproxy.SendMessageCallback;
import org.apache.inlong.sdk.dataproxy.SendResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_HOST;
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_PORT;
+
/**
* proxy client
*/
@@ -94,28 +95,29 @@ public class SenderManager {
localhost = jobConf.get(CommonConstants.PROXY_LOCAL_HOST, CommonConstants.DEFAULT_PROXY_LOCALHOST);
netTag = jobConf.get(CommonConstants.PROXY_NET_TAG, CommonConstants.DEFAULT_PROXY_NET_TAG);
isLocalVisit = jobConf.getBoolean(
- CommonConstants.PROXY_IS_LOCAL_VISIT, CommonConstants.DEFAULT_PROXY_IS_LOCAL_VISIT);
+ CommonConstants.PROXY_IS_LOCAL_VISIT, CommonConstants.DEFAULT_PROXY_IS_LOCAL_VISIT);
totalAsyncBufSize = jobConf
.getInt(
- CommonConstants.PROXY_TOTAL_ASYNC_PROXY_SIZE, CommonConstants.DEFAULT_PROXY_TOTAL_ASYNC_PROXY_SIZE);
+ CommonConstants.PROXY_TOTAL_ASYNC_PROXY_SIZE,
+ CommonConstants.DEFAULT_PROXY_TOTAL_ASYNC_PROXY_SIZE);
aliveConnectionNum = jobConf
.getInt(
- CommonConstants.PROXY_ALIVE_CONNECTION_NUM, CommonConstants.DEFAULT_PROXY_ALIVE_CONNECTION_NUM);
+ CommonConstants.PROXY_ALIVE_CONNECTION_NUM, CommonConstants.DEFAULT_PROXY_ALIVE_CONNECTION_NUM);
isCompress = jobConf.getBoolean(
- CommonConstants.PROXY_IS_COMPRESS, CommonConstants.DEFAULT_PROXY_IS_COMPRESS);
+ CommonConstants.PROXY_IS_COMPRESS, CommonConstants.DEFAULT_PROXY_IS_COMPRESS);
maxSenderPerGroup = jobConf.getInt(
- CommonConstants.PROXY_MAX_SENDER_PER_GROUP, CommonConstants.DEFAULT_PROXY_MAX_SENDER_PER_GROUP);
+ CommonConstants.PROXY_MAX_SENDER_PER_GROUP, CommonConstants.DEFAULT_PROXY_MAX_SENDER_PER_GROUP);
msgType = jobConf.getInt(CommonConstants.PROXY_MSG_TYPE, CommonConstants.DEFAULT_PROXY_MSG_TYPE);
maxSenderTimeout = jobConf.getInt(
- CommonConstants.PROXY_SENDER_MAX_TIMEOUT, CommonConstants.DEFAULT_PROXY_SENDER_MAX_TIMEOUT);
+ CommonConstants.PROXY_SENDER_MAX_TIMEOUT, CommonConstants.DEFAULT_PROXY_SENDER_MAX_TIMEOUT);
maxSenderRetry = jobConf.getInt(
- CommonConstants.PROXY_SENDER_MAX_RETRY, CommonConstants.DEFAULT_PROXY_SENDER_MAX_RETRY);
+ CommonConstants.PROXY_SENDER_MAX_RETRY, CommonConstants.DEFAULT_PROXY_SENDER_MAX_RETRY);
retrySleepTime = jobConf.getLong(
- CommonConstants.PROXY_RETRY_SLEEP, CommonConstants.DEFAULT_PROXY_RETRY_SLEEP);
+ CommonConstants.PROXY_RETRY_SLEEP, CommonConstants.DEFAULT_PROXY_RETRY_SLEEP);
isFile = jobConf.getBoolean(CommonConstants.PROXY_IS_FILE, CommonConstants.DEFAULT_IS_FILE);
taskPositionManager = TaskPositionManager.getTaskPositionManager();
semaphore = new Semaphore(jobConf.getInt(CommonConstants.PROXY_MESSAGE_SEMAPHORE,
- CommonConstants.DEFAULT_PROXY_MESSAGE_SEMAPHORE));
+ CommonConstants.DEFAULT_PROXY_MESSAGE_SEMAPHORE));
ioThreadNum = jobConf.getInt(CommonConstants.PROXY_CLIENT_IO_THREAD_NUM,
CommonConstants.DEFAULT_PROXY_CLIENT_IO_THREAD_NUM);
enableBusyWait = jobConf.getBoolean(CommonConstants.PROXY_CLIENT_ENABLE_BUSY_WAIT,
@@ -134,7 +136,7 @@ public class SenderManager {
/**
* Select by group.
*
- * @param group - inlong group id
+ * @param group inlong group id
* @return default message sender
*/
private DefaultMessageSender selectSender(String group) {
@@ -147,14 +149,14 @@ public class SenderManager {
semaphore.acquire(messageNum);
} catch (Exception e) {
LOGGER.error("acquire messageNum {} fail, current semaphore {}",
- messageNum, semaphore.availablePermits());
+ messageNum, semaphore.availablePermits());
}
}
/**
* sender
*
- * @param groupId - group id
+ * @param groupId group id
* @return DefaultMessageSender
*/
private DefaultMessageSender createMessageSender(String groupId) throws Exception {
@@ -176,7 +178,6 @@ public class SenderManager {
/**
* Add new sender for group id if max size is not satisfied.
- *
*/
public void addMessageSender() throws Exception {
List<DefaultMessageSender> tmpList = new ArrayList<>();
@@ -195,6 +196,7 @@ public class SenderManager {
* sender callback
*/
private class AgentSenderCallback implements SendMessageCallback {
+
private final int retry;
private final String groupId;
private final List<byte[]> bodyList;
@@ -203,7 +205,7 @@ public class SenderManager {
private final String jobId;
AgentSenderCallback(String jobId, String groupId, String streamId, List<byte[]> bodyList, int retry,
- long dataTime) {
+ long dataTime) {
this.retry = retry;
this.groupId = groupId;
this.streamId = streamId;
@@ -237,13 +239,13 @@ public class SenderManager {
/**
* Send message to proxy by batch, use message cache.
*
- * @param groupId - groupId
- * @param streamId - streamId
- * @param bodyList - body list
- * @param retry - retry time
+ * @param groupId groupId
+ * @param streamId streamId
+ * @param bodyList body list
+ * @param retry retry time
*/
public void sendBatchAsync(String jobId, String groupId, String streamId,
- List<byte[]> bodyList, int retry, long dataTime) {
+ List<byte[]> bodyList, int retry, long dataTime) {
if (retry > maxSenderRetry) {
LOGGER.warn("max retry reached, retry count is {}, sleep and send again", retry);
AgentUtils.silenceSleepInMs(retrySleepTime);
@@ -272,21 +274,21 @@ public class SenderManager {
/**
* Send message to proxy by batch, use message cache.
*
- * @param groupId - groupId
- * @param streamId - streamId
- * @param bodyList - body list
- * @param retry - retry time
+ * @param groupId groupId
+ * @param streamId streamId
+ * @param bodyList body list
+ * @param retry retry time
*/
public void sendBatchSync(String groupId, String streamId,
- List<byte[]> bodyList, int retry, long dataTime, Map<String, String> extraMap) {
+ List<byte[]> bodyList, int retry, long dataTime, Map<String, String> extraMap) {
if (retry > maxSenderRetry) {
LOGGER.warn("max retry reached, retry count is {}, sleep and send again", retry);
AgentUtils.silenceSleepInMs(retrySleepTime);
}
try {
selectSender(groupId).sendMessage(
- bodyList, groupId, streamId, dataTime, "",
- maxSenderTimeout, TimeUnit.SECONDS, extraMap
+ bodyList, groupId, streamId, dataTime, "",
+ maxSenderTimeout, TimeUnit.SECONDS, extraMap
);
semaphore.release(bodyList.size());
} catch (Exception exception) {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/BinlogSource.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/BinlogSource.java
index dd073eccf..c74a3726c 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/BinlogSource.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/BinlogSource.java
@@ -17,9 +17,6 @@
package org.apache.inlong.agent.plugin.sources;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.plugin.Reader;
import org.apache.inlong.agent.plugin.Source;
@@ -32,23 +29,27 @@ import org.apache.inlong.agent.utils.ConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class BinlogSource implements Source {
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
- private final SourceMetrics sourceMetrics;
+/**
+ * binlog source, split binlog source job into multi readers
+ */
+public class BinlogSource implements Source {
private static final Logger LOGGER = LoggerFactory.getLogger(TextFileSource.class);
-
private static final String BINLOG_SOURCE_TAG_NAME = "BinlogSourceMetric";
-
private static AtomicLong metricsIndex = new AtomicLong(0);
+ private final SourceMetrics sourceMetrics;
public BinlogSource() {
if (ConfigUtil.isPrometheusEnabled()) {
this.sourceMetrics = new SourcePrometheusMetrics(AgentUtils.getUniqId(
- BINLOG_SOURCE_TAG_NAME, metricsIndex.incrementAndGet()));
+ BINLOG_SOURCE_TAG_NAME, metricsIndex.incrementAndGet()));
} else {
this.sourceMetrics = new SourceJmxMetric(AgentUtils.getUniqId(
- BINLOG_SOURCE_TAG_NAME, metricsIndex.incrementAndGet()));
+ BINLOG_SOURCE_TAG_NAME, metricsIndex.incrementAndGet()));
}
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java
index b782b7385..87af9c4e6 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java
@@ -17,23 +17,7 @@
package org.apache.inlong.agent.plugin.sources;
-import static org.apache.inlong.agent.constant.JobConstants.DEFAULT_JOB_LINE_FILTER;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_ID;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_AUTO_COMMIT_OFFSET_RESET;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_BOOTSTRAP_SERVERS;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_GROUP_ID;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_OFFSET;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_PARTITION_OFFSET_DELIMITER;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_TOPIC;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_LINE_FILTER_PATTERN;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_OFFSET_DELIMITER;
import com.google.gson.Gson;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.agent.conf.JobProfile;
@@ -51,6 +35,27 @@ import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.inlong.agent.constant.JobConstants.DEFAULT_JOB_LINE_FILTER;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_ID;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_AUTO_COMMIT_OFFSET_RESET;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_BOOTSTRAP_SERVERS;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_GROUP_ID;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_OFFSET;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_PARTITION_OFFSET_DELIMITER;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_TOPIC;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_LINE_FILTER_PATTERN;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_OFFSET_DELIMITER;
+
+/**
+ * kafka source, split kafka source job into multi readers
+ */
public class KafkaSource implements Source {
public static final String JOB_KAFKA_AUTO_RESETE = "auto.offset.reset";
@@ -60,7 +65,7 @@ public class KafkaSource implements Source {
private static final String JOB_KAFKAJOB_WAIT_TIMEOUT = "job.kafkajob.wait.timeout";
private static final String KAFKA_COMMIT_AUTO = "enable.auto.commit";
private static final String KAFKA_DESERIALIZER_METHOD =
- "org.apache.kafka.common.serialization.ByteArrayDeserializer";
+ "org.apache.kafka.common.serialization.ByteArrayDeserializer";
private static final String KAFKA_KEY_DESERIALIZER = "key.deserializer";
private static final String KAFKA_VALUE_DESERIALIZER = "value.deserializer";
private static final String KAFKA_SESSION_TIMEOUT = "session.timeout.ms";
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/AbstractReader.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/AbstractReader.java
index b2b2291c1..7f5566d0e 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/AbstractReader.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/AbstractReader.java
@@ -35,6 +35,9 @@ import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STRE
import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_GROUP_ID;
import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_STREAM_ID;
+/**
+ * abstract reader, init reader and reader metrics
+ */
public abstract class AbstractReader implements Reader {
protected static PluginMetric readerMetric;
@@ -50,6 +53,11 @@ public abstract class AbstractReader implements Reader {
inlongStreamId = jobConf.get(PROXY_INLONG_STREAM_ID, DEFAULT_PROXY_INLONG_STREAM_ID);
}
+ /**
+ * init reader metrics
+ *
+ * @param tagName metric tagName
+ */
protected void intMetric(String tagName) {
String metricsIndexValue = String.valueOf(metricsIndex.getAndIncrement());
String label = Joiner.on(",").join(tagName, metricsIndexValue);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java
index db4f5c4cf..42b4302c9 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java
@@ -17,21 +17,11 @@
package org.apache.inlong.agent.plugin.sources.reader;
-import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_MAP_CAPACITY;
-import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_DATA;
-
import com.google.gson.Gson;
import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.relational.history.FileDatabaseHistory;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.conf.JobProfile;
@@ -47,9 +37,22 @@ import org.apache.inlong.common.reporpter.ConfigLogTypeEnum;
import org.apache.inlong.common.reporpter.StreamConfigLogMetric;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;
import org.slf4j.Logger;
-
import org.slf4j.LoggerFactory;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_MAP_CAPACITY;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_DATA;
+
+/**
+ * read binlog data
+ */
public class BinlogReader implements Reader {
private static final Logger LOGGER = LoggerFactory.getLogger(BinlogReader.class);
@@ -73,13 +76,13 @@ public class BinlogReader implements Reader {
private static final String JOB_DATABASE_HISTORY_MONITOR_DDL = "job.binlogJob.ddl";
private static final String JOB_DATABASE_PORT = "job.binlogJob.port";
private static final String JOB_DATABASE_QUEUE_SIZE = "job.binlogJob.queueSize";
-
+ private static final Gson gson = new Gson();
+ private final AgentConfiguration agentConf = AgentConfiguration.getAgentConf();
/**
* pair.left: table name
* pair.right: actual data
*/
private LinkedBlockingQueue<Pair<String, String>> binlogMessagesQueue;
-
private boolean finished = false;
private String userName;
private String password;
@@ -99,12 +102,9 @@ public class BinlogReader implements Reader {
private String offset;
private BinlogSnapshotBase binlogSnapshot;
private JobProfile jobProfile;
- private static final Gson gson = new Gson();
private boolean destroyed = false;
private boolean enableReportConfigLog;
private StreamConfigLogMetric streamConfigLogMetric;
- private final AgentConfiguration agentConf = AgentConfiguration.getAgentConf();
-
private String inlongGroupId;
private String inlongStreamId;
@@ -125,7 +125,7 @@ public class BinlogReader implements Reader {
Map<String, String> header = new HashMap<>(DEFAULT_MAP_CAPACITY);
header.put(PROXY_KEY_DATA, message.getKey());
return new DefaultMessage(message.getValue().getBytes(StandardCharsets.UTF_8),
- header);
+ header);
}
@Override
@@ -141,9 +141,9 @@ public class BinlogReader implements Reader {
serverTimeZone = jobConf.get(JOB_DATABASE_SERVER_TIME_ZONE, "");
offsetFlushIntervalMs = jobConf.get(JOB_DATABASE_STORE_OFFSET_INTERVAL_MS, "100000");
databaseStoreHistoryName = jobConf.get(JOB_DATABASE_STORE_HISTORY_FILENAME,
- tryToInitAndGetHistoryPath()) + "/history.dat" + jobConf.getInstanceId();
+ tryToInitAndGetHistoryPath()) + "/history.dat" + jobConf.getInstanceId();
offsetStoreFileName = jobConf.get(JOB_DATABASE_STORE_HISTORY_FILENAME,
- tryToInitAndGetHistoryPath()) + "/offset.dat" + jobConf.getInstanceId();
+ tryToInitAndGetHistoryPath()) + "/offset.dat" + jobConf.getInstanceId();
snapshotMode = jobConf.get(JOB_DATABASE_SNAPSHOT_MODE, "");
includeSchemaChanges = jobConf.get(JOB_DATABASE_INCLUDE_SCHEMA_CHANGES, "false");
historyMonitorDdl = jobConf.get(JOB_DATABASE_HISTORY_MONITOR_DDL, "false");
@@ -157,7 +157,7 @@ public class BinlogReader implements Reader {
enableReportConfigLog =
Boolean.parseBoolean(jobConf.get(StreamConfigLogMetric.CONFIG_LOG_REPORT_ENABLE,
- "true"));
+ "true"));
inlongGroupId = jobConf.get(CommonConstants.PROXY_INLONG_GROUP_ID,
CommonConstants.DEFAULT_PROXY_INLONG_GROUP_ID);
@@ -179,38 +179,38 @@ public class BinlogReader implements Reader {
Properties props = getEngineProps();
DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(
- io.debezium.engine.format.Json.class)
- .using(props)
- .notifying((records, committer) -> {
- try {
- for (ChangeEvent<String, String> record : records) {
- DebeziumFormat debeziumFormat = gson
- .fromJson(record.value(), DebeziumFormat.class);
- binlogMessagesQueue.put(Pair.of(debeziumFormat.getSource().getTable(),
- record.value()));
- committer.markProcessed(record);
+ io.debezium.engine.format.Json.class)
+ .using(props)
+ .notifying((records, committer) -> {
+ try {
+ for (ChangeEvent<String, String> record : records) {
+ DebeziumFormat debeziumFormat = gson
+ .fromJson(record.value(), DebeziumFormat.class);
+ binlogMessagesQueue.put(Pair.of(debeziumFormat.getSource().getTable(),
+ record.value()));
+ committer.markProcessed(record);
+ }
+ committer.markBatchFinished();
+ } catch (Exception e) {
+ LOGGER.error("parse binlog message error", e);
+ }
+
+ })
+ .using((success, message, error) -> {
+ if (!success) {
+ LOGGER.error("binlog job with jobConf {} has " + "error {}",
+ jobConf.getInstanceId(), message, error);
+ streamConfigLogMetric
+ .updateConfigLog(inlongGroupId, inlongStreamId, "DBConfig",
+ ConfigLogTypeEnum.ERROR, error == null ? "" : error.toString());
}
- committer.markBatchFinished();
- } catch (Exception e) {
- LOGGER.error("parse binlog message error", e);
- }
-
- })
- .using((success, message, error) -> {
- if (!success) {
- LOGGER.error("binlog job with jobConf {} has " + "error {}",
- jobConf.getInstanceId(), message, error);
- streamConfigLogMetric
- .updateConfigLog(inlongGroupId, inlongStreamId, "DBConfig",
- ConfigLogTypeEnum.ERROR, error == null ? "" : error.toString());
- }
- }).build();
+ }).build();
executor = Executors.newSingleThreadExecutor();
executor.execute(engine);
LOGGER.info("get initial snapshot of job {}, snapshot {}",
- jobConf.getInstanceId(), getSnapshot());
+ jobConf.getInstanceId(), getSnapshot());
}
private Properties getEngineProps() {
@@ -305,9 +305,9 @@ public class BinlogReader implements Reader {
private String tryToInitAndGetHistoryPath() {
String historyPath = agentConf.get(
- AgentConstants.AGENT_HISTORY_PATH, AgentConstants.DEFAULT_AGENT_HISTORY_PATH);
+ AgentConstants.AGENT_HISTORY_PATH, AgentConstants.DEFAULT_AGENT_HISTORY_PATH);
String parentPath = agentConf.get(
- AgentConstants.AGENT_HOME, AgentConstants.DEFAULT_AGENT_HOME);
+ AgentConstants.AGENT_HOME, AgentConstants.DEFAULT_AGENT_HOME);
return AgentUtils.makeDirsIfNotExist(historyPath, parentPath).getAbsolutePath();
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/KafkaReader.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/KafkaReader.java
index 1c314c0d3..fe28c79a6 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/KafkaReader.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/KafkaReader.java
@@ -17,25 +17,6 @@
package org.apache.inlong.agent.plugin.sources.reader;
-import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_GROUP_ID;
-import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_STREAM_ID;
-import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
-import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_BYTE_SPEED_LIMIT;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_OFFSET;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_PARTITION_OFFSET_DELIMITER;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_RECORD_SPEED_LIMIT;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_TOPIC;
-
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.agent.conf.JobProfile;
@@ -57,6 +38,29 @@ import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_PROXY_INLONG_STREAM_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_BYTE_SPEED_LIMIT;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_OFFSET;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_PARTITION_OFFSET_DELIMITER;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_RECORD_SPEED_LIMIT;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_KAFKA_TOPIC;
+
+/**
+ * read kafka data
+ */
public class KafkaReader<K, V> implements Reader {
public static final int NEVER_STOP_SIGN = -1;
@@ -96,9 +100,6 @@ public class KafkaReader<K, V> implements Reader {
/**
* init attribute
- *
- * @param consumer
- * @param paraMap
*/
public KafkaReader(KafkaConsumer<K, V> consumer, Map<String, String> paraMap) {
this.consumer = consumer;
@@ -222,6 +223,11 @@ public class KafkaReader<K, V> implements Reader {
return validators.stream().allMatch(v -> v.validate(new String(message)));
}
+ /**
+ * add specified pattern to validators
+ *
+ * @param pattern specified pattern
+ */
public void addPatternValidator(String pattern) {
if (pattern.isEmpty()) {
return;
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
index a42dd78b2..a627e7c98 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
@@ -17,18 +17,6 @@
package org.apache.inlong.agent.plugin.sources.reader;
-import static org.apache.inlong.agent.constant.JobConstants.DEFAULT_JOB_FILE_MAX_WAIT;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_FILE_MAX_WAIT;
-
-import java.io.File;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
-
import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.message.DefaultMessage;
@@ -41,14 +29,26 @@ import org.apache.inlong.agent.utils.AgentUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class TextFileReader extends AbstractReader {
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
- private static final Logger LOGGER = LoggerFactory.getLogger(TextFileReader.class);
+import static org.apache.inlong.agent.constant.JobConstants.DEFAULT_JOB_FILE_MAX_WAIT;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_FILE_MAX_WAIT;
- private static final String TEXT_FILE_READER_TAG_NAME = "AgentTextMetric";
+/**
+ * read file data
+ */
+public class TextFileReader extends AbstractReader {
public static final int NEVER_STOP_SIGN = -1;
-
+ private static final Logger LOGGER = LoggerFactory.getLogger(TextFileReader.class);
+ private static final String TEXT_FILE_READER_TAG_NAME = "AgentTextMetric";
private final File file;
private final int position;
@@ -197,6 +197,6 @@ public class TextFileReader extends AbstractReader {
}
AgentUtils.finallyClose(stream);
LOGGER.info("destroy reader with read {} num {}",
- streamMetric.getTagName(), streamMetric.getReadNum());
+ streamMetric.getTagName(), streamMetric.getReadNum());
}
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/snapshot/BinlogSnapshotBase.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/snapshot/BinlogSnapshotBase.java
index edc7d5438..a01deb19d 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/snapshot/BinlogSnapshotBase.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/snapshot/BinlogSnapshotBase.java
@@ -17,6 +17,9 @@
package org.apache.inlong.agent.plugin.sources.snapshot;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
@@ -26,18 +29,18 @@ import java.io.OutputStream;
import java.util.Base64;
import java.util.Base64.Decoder;
import java.util.Base64.Encoder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * binlog snapshot
+ */
public class BinlogSnapshotBase implements SnapshotBase {
public static final int BUFFER_SIZE = 1024;
public static final int START_OFFSET = 0;
private static final Logger log = LoggerFactory.getLogger(BinlogSnapshotBase.class);
- private File file;
private final Decoder decoder = Base64.getDecoder();
private final Encoder encoder = Base64.getEncoder();
-
+ private File file;
private byte[] offset;
public BinlogSnapshotBase(String filePath) {
@@ -54,6 +57,9 @@ public class BinlogSnapshotBase implements SnapshotBase {
public void close() {
}
+ /**
+ * load binlog offset from local file
+ */
public void load() {
try {
if (!file.exists()) {
@@ -75,6 +81,9 @@ public class BinlogSnapshotBase implements SnapshotBase {
}
}
+ /**
+ * save binlog offset to local file
+ */
public void save(String snapshot) {
byte[] bytes = decoder.decode(snapshot);
if (bytes.length != 0) {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/DirectoryTrigger.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/DirectoryTrigger.java
index 825c43fd0..944c68640 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/DirectoryTrigger.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/DirectoryTrigger.java
@@ -17,6 +17,16 @@
package org.apache.inlong.agent.plugin.trigger;
+import org.apache.inlong.agent.common.AbstractDaemon;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.conf.TriggerProfile;
+import org.apache.inlong.agent.constant.AgentConstants;
+import org.apache.inlong.agent.constant.JobConstants;
+import org.apache.inlong.agent.plugin.Trigger;
+import org.apache.inlong.agent.plugin.utils.PluginUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
@@ -34,19 +44,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
-import org.apache.inlong.agent.common.AbstractDaemon;
-import org.apache.inlong.agent.conf.JobProfile;
-import org.apache.inlong.agent.conf.TriggerProfile;
-import org.apache.inlong.agent.constant.AgentConstants;
-import org.apache.inlong.agent.constant.JobConstants;
-import org.apache.inlong.agent.plugin.Trigger;
-import org.apache.inlong.agent.plugin.utils.PluginUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
- * Watch directory, if new valid files are created, create
- * jobs correspondingly.
+ * Watch directory, if new valid files are created, create jobs correspondingly.
*/
public class DirectoryTrigger extends AbstractDaemon implements Trigger {
@@ -105,9 +105,9 @@ public class DirectoryTrigger extends AbstractDaemon implements Trigger {
/**
* register all sub-directory
*
- * @param entity - entity
- * @param path - path
- * @param tmpWatchers - watchers
+ * @param entity entity
+ * @param path path
+ * @param tmpWatchers watchers
*/
private void registerAllSubDir(PathPattern entity,
Path path,
@@ -126,9 +126,9 @@ public class DirectoryTrigger extends AbstractDaemon implements Trigger {
}
} else {
JobProfile copiedJobProfile = PluginUtils.copyJobProfile(profile,
- entity.getSuitTime(), path.toFile());
+ entity.getSuitTime(), path.toFile());
LOGGER.info("trigger {} generate job profile to read file {}",
- getTriggerProfile().getTriggerId(), path.toString());
+ getTriggerProfile().getTriggerId(), path.toString());
queue.offer(copiedJobProfile);
}
}
@@ -137,9 +137,9 @@ public class DirectoryTrigger extends AbstractDaemon implements Trigger {
/**
* if directory has created, then check whether directory is valid
*
- * @param entity - entity
- * @param watchKey - watch key
- * @param tmpWatchers - watchers
+ * @param entity entity
+ * @param watchKey watch key
+ * @param tmpWatchers watchers
*/
private void registerNewDir(PathPattern entity,
WatchKey watchKey,
@@ -219,11 +219,17 @@ public class DirectoryTrigger extends AbstractDaemon implements Trigger {
submitWorker(watchEventHandler());
}
+ /**
+ * register pathPattern into watchers
+ */
public void register(String pathPattern) throws IOException {
PathPattern entity = new PathPattern(pathPattern);
innerRegister(pathPattern, entity);
}
+ /**
+ * register pathPattern into watchers, with offset
+ */
public void register(String pathPattern, String offset) throws IOException {
PathPattern entity = new PathPattern(pathPattern, offset);
innerRegister(pathPattern, entity);
@@ -262,7 +268,7 @@ public class DirectoryTrigger extends AbstractDaemon implements Trigger {
public void init(TriggerProfile profile) throws IOException {
initWatchService();
interval = profile.getInt(
- AgentConstants.TRIGGER_CHECK_INTERVAL, AgentConstants.DEFAULT_TRIGGER_CHECK_INTERVAL);
+ AgentConstants.TRIGGER_CHECK_INTERVAL, AgentConstants.DEFAULT_TRIGGER_CHECK_INTERVAL);
this.profile = profile;
if (this.profile.hasKey(JobConstants.JOB_DIR_FILTER_PATTERN)) {
String pathPattern = this.profile.get(JobConstants.JOB_DIR_FILTER_PATTERN);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/PathPattern.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/PathPattern.java
index 96c76423d..7a068f46b 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/PathPattern.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/trigger/PathPattern.java
@@ -17,6 +17,12 @@
package org.apache.inlong.agent.plugin.trigger;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.inlong.agent.plugin.filter.DateFormatRegex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
@@ -27,12 +33,6 @@ import java.util.HashSet;
import java.util.Set;
import java.util.stream.Stream;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-import org.apache.inlong.agent.plugin.filter.DateFormatRegex;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
/**
* path pattern for file filter.
*/
@@ -62,8 +62,6 @@ public class PathPattern {
/**
* find last existing path by pattern.
- * @param watchDir
- * @return
*/
private String findRoot(String watchDir) {
Path currentPath = Paths.get(watchDir);
@@ -81,11 +79,9 @@ public class PathPattern {
/**
* walk all suitable files under directory.
- * @param collectResult
- * @param maxNum
*/
private void walkAllSuitableFiles(File dirPath, final Collection<File> collectResult,
- int maxNum) throws IOException {
+ int maxNum) throws IOException {
if (collectResult.size() > maxNum) {
LOGGER.warn("max num of files is {}, please check", maxNum);
return;
@@ -109,12 +105,9 @@ public class PathPattern {
/**
* walk root directory
- * @param collectResult
- * @param maxNum
- * @throws IOException
*/
public void walkAllSuitableFiles(final Collection<File> collectResult,
- int maxNum) throws IOException {
+ int maxNum) throws IOException {
walkAllSuitableFiles(new File(rootDir), collectResult, maxNum);
}
@@ -129,7 +122,7 @@ public class PathPattern {
/**
* whether path is suitable
*
- * @param pathStr - pathString
+ * @param pathStr pathString
* @return true if suit else false.
*/
public boolean suitForWatch(String pathStr) {
@@ -150,7 +143,6 @@ public class PathPattern {
/**
* when a new file is found, update regex since time may change.
- * @param
*/
public void updateDateFormatRegex() {
dateFormatRegex.setRegexWithCurrentTime(this.watchDir);
@@ -158,7 +150,6 @@ public class PathPattern {
/**
* when job is retry job, the time for searching file should be specified.
- * @param time
*/
public void updateDateFormatRegex(String time) {
dateFormatRegex.setRegexWithTime(this.watchDir, time);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/BinlogTimeConverter.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/BinlogTimeConverter.java
index 07dedc690..1bad68d4e 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/BinlogTimeConverter.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/BinlogTimeConverter.java
@@ -19,6 +19,10 @@ package org.apache.inlong.agent.plugin.utils;
import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.time.DateTimeException;
import java.time.Duration;
import java.time.LocalDate;
@@ -26,14 +30,13 @@ import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
-import org.apache.kafka.connect.data.SchemaBuilder;
-
import java.time.format.DateTimeFormatter;
import java.util.Properties;
import java.util.function.Consumer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * binlog time converter utils
+ */
public class BinlogTimeConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
private static final Logger LOGGER = LoggerFactory.getLogger(PluginUtils.class);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/PluginUtils.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/PluginUtils.java
index 3da24dd5c..6855dc9fd 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/PluginUtils.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/PluginUtils.java
@@ -17,23 +17,6 @@
package org.apache.inlong.agent.plugin.utils;
-import static org.apache.inlong.agent.constant.CommonConstants.AGENT_COLON;
-import static org.apache.inlong.agent.constant.CommonConstants.AGENT_NIX_OS;
-import static org.apache.inlong.agent.constant.CommonConstants.AGENT_NUX_OS;
-import static org.apache.inlong.agent.constant.CommonConstants.AGENT_OS_NAME;
-import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_FILE_MAX_NUM;
-import static org.apache.inlong.agent.constant.CommonConstants.FILE_MAX_NUM;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_DIR_FILTER_PATTERN;
-import static org.apache.inlong.agent.constant.JobConstants.JOB_RETRY_TIME;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.NetworkInterface;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Enumeration;
-import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.conf.TriggerProfile;
@@ -44,6 +27,24 @@ import org.apache.pulsar.client.api.CompressionType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Enumeration;
+import java.util.List;
+
+import static org.apache.inlong.agent.constant.CommonConstants.AGENT_COLON;
+import static org.apache.inlong.agent.constant.CommonConstants.AGENT_NIX_OS;
+import static org.apache.inlong.agent.constant.CommonConstants.AGENT_NUX_OS;
+import static org.apache.inlong.agent.constant.CommonConstants.AGENT_OS_NAME;
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_FILE_MAX_NUM;
+import static org.apache.inlong.agent.constant.CommonConstants.FILE_MAX_NUM;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_DIR_FILTER_PATTERN;
+import static org.apache.inlong.agent.constant.JobConstants.JOB_RETRY_TIME;
+
/**
* Utils for plugin package.
*/
@@ -54,21 +55,26 @@ public class PluginUtils {
/**
* convert string compress type to enum compress type
- * @param type
- * @return
*/
public static CompressionType convertType(String type) {
switch (type) {
- case "lz4": return CompressionType.LZ4;
- case "zlib": return CompressionType.ZLIB;
- case "zstd": return CompressionType.ZSTD;
- case "snappy": return CompressionType.SNAPPY;
+ case "lz4":
+ return CompressionType.LZ4;
+ case "zlib":
+ return CompressionType.ZLIB;
+ case "zstd":
+ return CompressionType.ZSTD;
+ case "snappy":
+ return CompressionType.SNAPPY;
case "none":
default:
return CompressionType.NONE;
}
}
+ /**
+ * scan and return files based on job dir conf
+ */
public static Collection<File> findSuitFiles(JobProfile jobConf) {
String dirPattern = jobConf.get(JOB_DIR_FILTER_PATTERN);
LOGGER.info("start to find files with dir pattern {}", dirPattern);
@@ -87,19 +93,20 @@ public class PluginUtils {
/**
* if the job is retry job, the date is determined
- * @param jobConf
- * @param pattern
*/
public static void updateRetryTime(JobProfile jobConf, PathPattern pattern) {
if (jobConf.hasKey(JOB_RETRY_TIME)) {
LOGGER.info("job {} is retry job with specific time, update file time to {}"
- + "", jobConf.toJsonStr(), jobConf.get(JOB_RETRY_TIME));
+ + "", jobConf.toJsonStr(), jobConf.get(JOB_RETRY_TIME));
pattern.updateDateFormatRegex(jobConf.get(JOB_RETRY_TIME));
}
}
+ /**
+ * convert TriggerProfile to JobProfile
+ */
public static JobProfile copyJobProfile(TriggerProfile triggerProfile, String dataTime,
- File pendingFile) {
+ File pendingFile) {
JobProfile copiedProfile = TriggerProfile.parseJsonStr(triggerProfile.toJsonStr());
String md5 = AgentUtils.getFileMd5(pendingFile);
copiedProfile.set(pendingFile.getAbsolutePath() + ".md5", md5);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PatternValidator.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PatternValidator.java
index a7407feda..1f6671e38 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PatternValidator.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PatternValidator.java
@@ -19,14 +19,17 @@ package org.apache.inlong.agent.plugin.validator;
import org.apache.inlong.agent.plugin.Validator;
+/**
+ * validate whether a message contains specified content
+ */
public class PatternValidator implements Validator {
+ private String pattern;
+
public PatternValidator(String pattern) {
this.pattern = pattern;
}
- private String pattern;
-
@Override
public boolean validate(String messageLine) {
return messageLine.contains(pattern);
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PrefixValidator.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PrefixValidator.java
index 0163acade..e20a6fffb 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PrefixValidator.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/validator/PrefixValidator.java
@@ -19,14 +19,17 @@ package org.apache.inlong.agent.plugin.validator;
import org.apache.inlong.agent.plugin.Validator;
+/**
+ * validate whether a message has a specified prefix
+ */
public class PrefixValidator implements Validator {
+ private String prefix;
+
public PrefixValidator(String prefix) {
this.prefix = prefix;
}
- private String prefix;
-
@Override
public boolean validate(String messageLine) {
return messageLine.startsWith(prefix);