You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2024/05/29 11:59:06 UTC
(inlong) branch master updated: [INLONG-10302][Agent] Add an interface for limiting the number of instances obtained (#10303)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 12efb53dae [INLONG-10302][Agent] Add an interface for limiting the number of instances obtained (#10303)
12efb53dae is described below
commit 12efb53dae42dc38075746a5c062ea69dd6bb7fd
Author: justinwwhuang <hw...@163.com>
AuthorDate: Wed May 29 19:59:01 2024 +0800
[INLONG-10302][Agent] Add an interface for limiting the number of instances obtained (#10303)
---
.../main/java/org/apache/inlong/agent/plugin/task/AbstractTask.java | 5 +++--
.../main/java/org/apache/inlong/agent/plugin/task/KafkaTask.java | 6 ++++++
.../main/java/org/apache/inlong/agent/plugin/task/MongoDBTask.java | 6 ++++++
.../main/java/org/apache/inlong/agent/plugin/task/PulsarTask.java | 6 ++++++
.../java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java | 5 +++++
5 files changed, 26 insertions(+), 2 deletions(-)
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/AbstractTask.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/AbstractTask.java
index a6d8d03482..acf8287817 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/AbstractTask.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/AbstractTask.java
@@ -19,7 +19,6 @@ package org.apache.inlong.agent.plugin.task;
import org.apache.inlong.agent.conf.InstanceProfile;
import org.apache.inlong.agent.conf.TaskProfile;
-import org.apache.inlong.agent.constant.TaskConstants;
import org.apache.inlong.agent.core.instance.ActionType;
import org.apache.inlong.agent.core.instance.InstanceAction;
import org.apache.inlong.agent.core.instance.InstanceManager;
@@ -58,7 +57,7 @@ public abstract class AbstractTask extends Task {
this.taskProfile = taskProfile;
this.basicDb = basicDb;
auditVersion = Long.parseLong(taskProfile.get(TASK_AUDIT_VERSION));
- instanceManager = new InstanceManager(taskProfile.getTaskId(), taskProfile.getInt(TaskConstants.FILE_MAX_NUM),
+ instanceManager = new InstanceManager(taskProfile.getTaskId(), getInstanceLimit(),
basicDb, taskManager.getTaskDb());
try {
instanceManager.start();
@@ -69,6 +68,8 @@ public abstract class AbstractTask extends Task {
initOK = true;
}
+ protected abstract int getInstanceLimit();
+
protected abstract void initTask();
protected void releaseTask() {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/KafkaTask.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/KafkaTask.java
index f83104e8a7..e7058c9929 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/KafkaTask.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/KafkaTask.java
@@ -36,10 +36,16 @@ public class KafkaTask extends AbstractTask {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaTask.class);
public static final String DEFAULT_KAFKA_INSTANCE = "org.apache.inlong.agent.plugin.instance.KafkaInstance";
+ public static final int DEFAULT_INSTANCE_LIMIT = 1;
private boolean isAdded = false;
private String topic;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMddHH");
+ @Override
+ protected int getInstanceLimit() {
+ return DEFAULT_INSTANCE_LIMIT;
+ }
+
@Override
protected void initTask() {
LOGGER.info("kafka commonInit: {}", taskProfile.toJsonStr());
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/MongoDBTask.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/MongoDBTask.java
index 2a022ddb78..01f97029b2 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/MongoDBTask.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/MongoDBTask.java
@@ -35,10 +35,16 @@ public class MongoDBTask extends AbstractTask {
private static final Logger LOGGER = LoggerFactory.getLogger(MongoDBTask.class);
public static final String DEFAULT_MONGODB_INSTANCE = "org.apache.inlong.agent.plugin.instance.MongoDBInstance";
+ public static final int DEFAULT_INSTANCE_LIMIT = 1;
private boolean isAdded = false;
private String collection;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMddHH");
+ @Override
+ protected int getInstanceLimit() {
+ return DEFAULT_INSTANCE_LIMIT;
+ }
+
@Override
protected void initTask() {
LOGGER.info("mongoDB commonInit: {}", taskProfile.toJsonStr());
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/PulsarTask.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/PulsarTask.java
index a105c45774..62b32dcbca 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/PulsarTask.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/PulsarTask.java
@@ -38,6 +38,7 @@ public class PulsarTask extends AbstractTask {
private static final Logger LOGGER = LoggerFactory.getLogger(PulsarTask.class);
public static final String DEFAULT_PULSAR_INSTANCE = "org.apache.inlong.agent.plugin.instance.PulsarInstance";
+ public static final int DEFAULT_INSTANCE_LIMIT = 1;
private boolean isAdded = false;
private String tenant;
private String namespace;
@@ -45,6 +46,11 @@ public class PulsarTask extends AbstractTask {
private String instanceId;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMddHH");
+ @Override
+ protected int getInstanceLimit() {
+ return DEFAULT_INSTANCE_LIMIT;
+ }
+
@Override
protected void initTask() {
LOGGER.info("pulsar commonInit: {}", taskProfile.toJsonStr());
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
index 4e6493c02a..44495b6059 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
@@ -87,6 +87,11 @@ public class LogFileTask extends AbstractTask {
private volatile long coreThreadUpdateTime = 0;
private BlockingQueue<InstanceProfile> instanceQueue;
+ @Override
+ protected int getInstanceLimit() {
+ return taskProfile.getInt(TaskConstants.FILE_MAX_NUM);
+ }
+
@Override
protected void initTask() {
instanceQueue = new LinkedBlockingQueue<>(INSTANCE_QUEUE_CAPACITY);