You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/09/09 08:43:43 UTC
[iotdb] branch master updated: [IOTDB-4326] Implementation of Create Trigger on DataNode (#7257)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new ecee938113 [IOTDB-4326] Implementation of Create Trigger on DataNode (#7257)
ecee938113 is described below
commit ecee938113e485e75aedd30a41cca7705c66b020
Author: Liao Lanyu <10...@users.noreply.github.com>
AuthorDate: Fri Sep 9 16:43:37 2022 +0800
[IOTDB-4326] Implementation of Create Trigger on DataNode (#7257)
---
.../iotdb/confignode/conf/ConfigNodeConfig.java | 2 +-
.../apache/iotdb/commons/conf/IoTDBConstant.java | 4 +-
.../commons/executable/ExecutableManager.java | 54 ++++++
.../org/apache/iotdb/commons/path/PartialPath.java | 8 +
.../apache/iotdb/commons/service/ServiceType.java | 1 +
.../iotdb/commons/trigger/TriggerInformation.java | 12 ++
.../apache/iotdb/commons/trigger/TriggerTable.java | 12 ++
...eption.java => TriggerJarToLargeException.java} | 6 +-
...eption.java => TriggerManagementException.java} | 6 +-
...nService.java => TriggerExecutableManager.java} | 60 +++----
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 25 ++-
.../config/executor/ClusterConfigTaskExecutor.java | 39 +++--
.../config/executor/IConfigTaskExecutor.java | 13 +-
.../executor/StandaloneConfigTaskExecutor.java | 15 +-
.../config/metadata/CreateTriggerTask.java | 27 +--
.../java/org/apache/iotdb/db/service/DataNode.java | 17 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 39 ++++-
.../iotdb/db/trigger/executor/TriggerExecutor.java | 57 +++++++
.../trigger/service/TriggerManagementService.java | 189 +++++++++++++++++++++
.../src/main/thrift/confignode.thrift | 10 +-
thrift/src/main/thrift/datanode.thrift | 41 +++--
21 files changed, 506 insertions(+), 131 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
index af13b0e7c7..28596863dd 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
@@ -102,7 +102,7 @@ public class ConfigNodeConfig {
/** External temporary lib directory for storing downloaded JAR files */
private String temporaryLibDir =
- IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.TMP_FOLDER_NAME;
+ IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.UDF_TMP_FOLDER_NAME;
/** Time partition interval in seconds */
private long timePartitionInterval = 86400;
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java b/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
index 66e8c157df..5d2c219e60 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
@@ -196,8 +196,10 @@ public class IoTDBConstant {
public static final String TRACING_LOG = "tracing.txt";
public static final String EXT_FOLDER_NAME = "ext";
public static final String UDF_FOLDER_NAME = "udf";
- public static final String TMP_FOLDER_NAME = "temporary";
+ public static final String UDF_TMP_FOLDER_NAME = "udf_temporary";
public static final String TRIGGER_FOLDER_NAME = "trigger";
+
+ public static final String TRIGGER_TMP_FOLDER_NAME = "trigger_temporary";
public static final String MQTT_FOLDER_NAME = "mqtt";
public static final String WAL_FOLDER_NAME = "wal";
public static final String EXT_PIPE_FOLDER_NAME = "extPipe";
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/executable/ExecutableManager.java b/node-commons/src/main/java/org/apache/iotdb/commons/executable/ExecutableManager.java
index 7022416d14..4be42361ab 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/executable/ExecutableManager.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/executable/ExecutableManager.java
@@ -19,20 +19,31 @@
package org.apache.iotdb.commons.executable;
+import org.apache.iotdb.commons.trigger.exception.TriggerJarToLargeException;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
public class ExecutableManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExecutableManager.class);
+
protected final String temporaryLibRoot;
protected final String libRoot;
@@ -112,4 +123,47 @@ public class ExecutableManager {
public String getDirStringByName(String name) {
return libRoot + File.separator + name + File.separator;
}
+
+ /////////////////////////////////////////////////////////////////////////////////////////////////
+ // transfer jar file to bytebuffer for thrift
+ /////////////////////////////////////////////////////////////////////////////////////////////////
+
+ public static ByteBuffer transferToBytebuffer(String filePath) throws IOException {
+ try (FileChannel fileChannel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ)) {
+ long size = fileChannel.size();
+ if (size > Integer.MAX_VALUE) {
+ // Max length of Thrift Binary is Integer.MAX_VALUE bytes.
+ throw new TriggerJarToLargeException(
+ String.format("Size of file exceed %d bytes", Integer.MAX_VALUE));
+ }
+ ByteBuffer byteBuffer = ByteBuffer.allocate((int) size);
+ fileChannel.read(byteBuffer);
+ return byteBuffer;
+ } catch (Exception e) {
+ LOGGER.warn(
+ "Error occurred during transferring file{} to ByteBuffer, the cause is {}",
+ filePath,
+ e.getMessage());
+ throw e;
+ }
+ }
+
+ /**
+ * @param byteBuffer jar data
+ * @param fileName The name of the file. Absolute Path will be libRoot + File_Separator + fileName
+ */
+ public void writeToLibDir(ByteBuffer byteBuffer, String fileName) throws IOException {
+ String destination = this.libRoot + File.separator + fileName;
+ Path path = Paths.get(destination);
+ Files.deleteIfExists(path);
+ try (FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.WRITE)) {
+ fileChannel.write(byteBuffer);
+ } catch (IOException e) {
+ LOGGER.warn(
+ "Error occurred during writing bytebuffer to {} , the cause is {}",
+ destination,
+ e.getMessage());
+ throw e;
+ }
+ }
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java b/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java
index 9d0b432797..bec99ab45f 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
@@ -598,6 +599,13 @@ public class PartialPath extends Path implements Comparable<Path>, Cloneable {
return new PartialPath(this.getNodes().clone());
}
+ public ByteBuffer serialize() throws IOException {
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+ serialize(outputStream);
+ return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+ }
+
@Override
public void serialize(ByteBuffer byteBuffer) {
PathType.Partial.serialize(byteBuffer);
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java b/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
index f9bdc39eb1..239ef700be 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
@@ -46,6 +46,7 @@ public enum ServiceType {
UDF_EXECUTABLE_MANAGER_SERVICE("UDF Executable Manager Service", ""),
TRIGGER_CLASSLOADER_MANAGER_SERVICE("Trigger ClassLoader Manager Service", ""),
TRIGGER_REGISTRATION_SERVICE("Trigger Registration Service", ""),
+ TRIGGER_EXECUTABLE_MANAGER_SERVICE("Trigger Executable Manager Service", ""),
TEMPORARY_QUERY_DATA_FILE_SERVICE("Temporary Query Data File Service", ""),
TRIGGER_REGISTRATION_SERVICE_OLD("Old Standalone Trigger Registration Service", ""),
CACHE_HIT_RATIO_DISPLAY_SERVICE(
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerInformation.java b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerInformation.java
index 3b31fb484a..bc66dd28b8 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerInformation.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerInformation.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Map;
/** This Class used to save the specific information of one Trigger. */
public class TriggerInformation {
@@ -36,6 +37,9 @@ public class TriggerInformation {
private String triggerName;
private String className;
private String jarName;
+
+ private Map<String, String> attributes;
+
private TTriggerState triggerState;
/** indicate this Trigger is Stateful or Stateless */
@@ -51,6 +55,7 @@ public class TriggerInformation {
String triggerName,
String className,
String jarName,
+ Map<String, String> attributes,
TTriggerState triggerState,
boolean isStateful,
TDataNodeLocation dataNodeLocation) {
@@ -58,6 +63,7 @@ public class TriggerInformation {
this.triggerName = triggerName;
this.className = className;
this.jarName = jarName;
+ this.attributes = attributes;
this.triggerState = triggerState;
this.isStateful = isStateful;
this.dataNodeLocation = dataNodeLocation;
@@ -75,6 +81,7 @@ public class TriggerInformation {
ReadWriteIOUtils.write(triggerName, outputStream);
ReadWriteIOUtils.write(className, outputStream);
ReadWriteIOUtils.write(jarName, outputStream);
+ ReadWriteIOUtils.write(attributes, outputStream);
ReadWriteIOUtils.write(triggerState.getValue(), outputStream);
ReadWriteIOUtils.write(isStateful, outputStream);
if (isStateful) {
@@ -88,6 +95,7 @@ public class TriggerInformation {
triggerInformation.triggerName = ReadWriteIOUtils.readString(byteBuffer);
triggerInformation.className = ReadWriteIOUtils.readString(byteBuffer);
triggerInformation.jarName = ReadWriteIOUtils.readString(byteBuffer);
+ triggerInformation.attributes = ReadWriteIOUtils.readMap(byteBuffer);
triggerInformation.triggerState =
TTriggerState.findByValue(ReadWriteIOUtils.readInt(byteBuffer));
boolean isStateful = ReadWriteIOUtils.readBool(byteBuffer);
@@ -123,6 +131,10 @@ public class TriggerInformation {
this.className = className;
}
+ public Map<String, String> getAttributes() {
+ return attributes;
+ }
+
public String getJarName() {
return jarName;
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerTable.java b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerTable.java
index 574222e255..ffba5e9e05 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerTable.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/TriggerTable.java
@@ -31,6 +31,10 @@ import java.util.Map;
public class TriggerTable {
private final Map<String, TriggerInformation> triggerTable;
+ // todo: Maintain a PatternTree: PathPattern -> List<String> triggerNames
+ // Given a PathPattern, return the triggerNames of triggers whose PathPatterns match the given
+ // one.
+
public TriggerTable() {
triggerTable = new HashMap<>();
}
@@ -49,6 +53,14 @@ public class TriggerTable {
triggerTable.remove(triggerName);
}
+ public boolean containsTrigger(String triggerName) {
+ return triggerTable.containsKey(triggerName);
+ }
+
+ public void activeTrigger(String triggerName) {
+ triggerTable.get(triggerName).setTriggerState(TTriggerState.ACTIVE);
+ }
+
// for showTrigger
public Map<String, TTriggerState> getAllTriggerStates() {
Map<String, TTriggerState> allTriggerStates = new HashMap<>(triggerTable.size());
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerRegistrationException.java b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerJarToLargeException.java
similarity index 82%
copy from node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerRegistrationException.java
copy to node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerJarToLargeException.java
index 34a18dc6c2..82705274bf 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerRegistrationException.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerJarToLargeException.java
@@ -19,12 +19,12 @@
package org.apache.iotdb.commons.trigger.exception;
-public class TriggerRegistrationException extends RuntimeException {
- public TriggerRegistrationException(String message) {
+public class TriggerJarToLargeException extends RuntimeException {
+ public TriggerJarToLargeException(String message) {
super(message);
}
- public TriggerRegistrationException(String message, Throwable cause) {
+ public TriggerJarToLargeException(String message, Throwable cause) {
super(message, cause);
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerRegistrationException.java b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerManagementException.java
similarity index 82%
rename from node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerRegistrationException.java
rename to node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerManagementException.java
index 34a18dc6c2..a60782478f 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerRegistrationException.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/exception/TriggerManagementException.java
@@ -19,12 +19,12 @@
package org.apache.iotdb.commons.trigger.exception;
-public class TriggerRegistrationException extends RuntimeException {
- public TriggerRegistrationException(String message) {
+public class TriggerManagementException extends RuntimeException {
+ public TriggerManagementException(String message) {
super(message);
}
- public TriggerRegistrationException(String message, Throwable cause) {
+ public TriggerManagementException(String message, Throwable cause) {
super(message, cause);
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/service/TriggerRegistrationService.java b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/service/TriggerExecutableManager.java
similarity index 55%
rename from node-commons/src/main/java/org/apache/iotdb/commons/trigger/service/TriggerRegistrationService.java
rename to node-commons/src/main/java/org/apache/iotdb/commons/trigger/service/TriggerExecutableManager.java
index 68e7e2074b..25f925b80c 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/trigger/service/TriggerRegistrationService.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/trigger/service/TriggerExecutableManager.java
@@ -20,45 +20,28 @@
package org.apache.iotdb.commons.trigger.service;
import org.apache.iotdb.commons.exception.StartupException;
+import org.apache.iotdb.commons.executable.ExecutableManager;
+import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.service.IService;
import org.apache.iotdb.commons.service.ServiceType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.locks.ReentrantLock;
-
-public class TriggerRegistrationService implements IService {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(TriggerRegistrationService.class);
-
- private final ReentrantLock registrationLock;
-
- private TriggerRegistrationService() {
- this.registrationLock = new ReentrantLock();
- }
-
- public void acquireRegistrationLock() {
- registrationLock.lock();
+public class TriggerExecutableManager extends ExecutableManager implements IService {
+ private TriggerExecutableManager(String temporaryLibRoot, String triggerLibRoot) {
+ super(temporaryLibRoot, triggerLibRoot);
}
- public void releaseRegistrationLock() {
- registrationLock.unlock();
- }
-
- // todo: implementation
- public void register() {
- // validate before registering
- // add to triggerTable and set inactive
- // throw exception if registered
- };
-
- public void activeTrigger(String triggerName) {
- // active trigger in table
- };
-
+ /////////////////////////////////////////////////////////////////////////////////////////////////
+ // IService
+ /////////////////////////////////////////////////////////////////////////////////////////////////
@Override
- public void start() throws StartupException {}
+ public void start() throws StartupException {
+ try {
+ SystemFileFactory.INSTANCE.makeDirIfNecessary(temporaryLibRoot);
+ SystemFileFactory.INSTANCE.makeDirIfNecessary(libRoot);
+ } catch (Exception e) {
+ throw new StartupException(e);
+ }
+ }
@Override
public void stop() {
@@ -67,23 +50,24 @@ public class TriggerRegistrationService implements IService {
@Override
public ServiceType getID() {
- return ServiceType.TRIGGER_REGISTRATION_SERVICE;
+ return ServiceType.TRIGGER_EXECUTABLE_MANAGER_SERVICE;
}
/////////////////////////////////////////////////////////////////////////////////////////////////
// singleton instance holder
/////////////////////////////////////////////////////////////////////////////////////////////////
- private static TriggerRegistrationService INSTANCE = null;
+ private static TriggerExecutableManager INSTANCE = null;
- public static synchronized TriggerRegistrationService setupAndGetInstance() {
+ public static synchronized TriggerExecutableManager setupAndGetInstance(
+ String temporaryLibRoot, String triggerLibRoot) {
if (INSTANCE == null) {
- INSTANCE = new TriggerRegistrationService();
+ INSTANCE = new TriggerExecutableManager(temporaryLibRoot, triggerLibRoot);
}
return INSTANCE;
}
- public static TriggerRegistrationService getInstance() {
+ public static TriggerExecutableManager getInstance() {
return INSTANCE;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 962aed8185..c858d356f8 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -257,14 +257,18 @@ public class IoTDBConfig {
private String udfDir =
IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.UDF_FOLDER_NAME;
- /** External temporary lib directory for storing downloaded JAR files */
- private String temporaryLibDir =
- IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.TMP_FOLDER_NAME;
+ /** External temporary lib directory for storing downloaded udf JAR files */
+ private String udfTemporaryLibDir =
+ IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.UDF_TMP_FOLDER_NAME;
/** External lib directory for trigger, stores user-uploaded JAR files */
private String triggerDir =
IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.TRIGGER_FOLDER_NAME;
+ /** External temporary lib directory for storing downloaded trigger JAR files */
+ private String triggerTemporaryLibDir =
+ IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.UDF_TMP_FOLDER_NAME;
+
/** External lib directory for ext Pipe plugins, stores user-defined JAR files */
private String extPipeDir =
IoTDBConstant.EXT_FOLDER_NAME + File.separator + IoTDBConstant.EXT_PIPE_FOLDER_NAME;
@@ -1098,8 +1102,9 @@ public class IoTDBConfig {
indexRootFolder = addHomeDir(indexRootFolder);
extDir = addHomeDir(extDir);
udfDir = addHomeDir(udfDir);
- temporaryLibDir = addHomeDir(temporaryLibDir);
+ udfTemporaryLibDir = addHomeDir(udfTemporaryLibDir);
triggerDir = addHomeDir(triggerDir);
+ triggerTemporaryLibDir = addHomeDir(triggerTemporaryLibDir);
mqttDir = addHomeDir(mqttDir);
externalPropertiesLoaderDir = addHomeDir(externalPropertiesLoaderDir);
externalLimiterDir = addHomeDir(externalLimiterDir);
@@ -1316,14 +1321,14 @@ public class IoTDBConfig {
return udfDir;
}
- public String getTemporaryLibDir() {
- return temporaryLibDir;
- }
-
public void setUdfDir(String udfDir) {
this.udfDir = udfDir;
}
+ public String getUdfTemporaryLibDir() {
+ return udfTemporaryLibDir;
+ }
+
public String getTriggerDir() {
return triggerDir;
}
@@ -1332,6 +1337,10 @@ public class IoTDBConfig {
this.triggerDir = triggerDir;
}
+ public String getTriggerTemporaryLibDir() {
+ return triggerTemporaryLibDir;
+ }
+
public String getMqttDir() {
return mqttDir;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 0f770e6509..cc1ee8e967 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -26,9 +26,8 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.executable.ExecutableManager;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.trigger.enums.TriggerEvent;
-import org.apache.iotdb.commons.trigger.enums.TriggerType;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateTriggerReq;
@@ -64,6 +63,7 @@ import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowNodes
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowPathSetTemplateTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowSchemaTemplateTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
@@ -255,22 +255,35 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
@Override
public SettableFuture<ConfigTaskResult> createTrigger(
- String triggerName,
- String className,
- String jarPath,
- boolean usingURI,
- TriggerEvent triggerEvent,
- TriggerType triggerType,
- PartialPath pathPattern) {
+ CreateTriggerStatement createTriggerStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
try (ConfigNodeClient client =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- // todo : createTriggerReq
- final TSStatus executionStatus =
- client.createTrigger(new TCreateTriggerReq(triggerName, null, null));
+ TCreateTriggerReq tCreateTriggerReq =
+ new TCreateTriggerReq(
+ createTriggerStatement.getTriggerName(),
+ createTriggerStatement.getClassName(),
+ createTriggerStatement.getJarPath(),
+ createTriggerStatement.isUsingURI(),
+ createTriggerStatement.getTriggerEvent().getId(),
+ createTriggerStatement.getTriggerType().getId(),
+ createTriggerStatement.getPathPattern().serialize(),
+ createTriggerStatement.getAttributes());
+
+ if (!createTriggerStatement.isUsingURI()) {
+ // If jarPath is a file path, we transfer it to ByteBuffer and send it to ConfigNode.
+ tCreateTriggerReq.setJarFile(
+ ExecutableManager.transferToBytebuffer(createTriggerStatement.getJarPath()));
+ }
+
+ final TSStatus executionStatus = client.createTrigger(tCreateTriggerReq);
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
- LOGGER.error("[{}] Failed to create trigger {}.", executionStatus, triggerName);
+ LOGGER.error(
+ "[{}] Failed to create trigger {}. TSStatus is {}",
+ executionStatus,
+ createTriggerStatement.getTriggerName(),
+ executionStatus.message);
future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
index 861f42277c..90750c2526 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -21,11 +21,9 @@ package org.apache.iotdb.db.mpp.plan.execution.config.executor;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.commons.cluster.NodeStatus;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.trigger.enums.TriggerEvent;
-import org.apache.iotdb.commons.trigger.enums.TriggerType;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
@@ -70,14 +68,7 @@ public interface IConfigTaskExecutor {
SettableFuture<ConfigTaskResult> dropFunction(String udfName);
- SettableFuture<ConfigTaskResult> createTrigger(
- String triggerName,
- String className,
- String jarPath,
- boolean usingURI,
- TriggerEvent triggerEvent,
- TriggerType triggerType,
- PartialPath pathPattern);
+ SettableFuture<ConfigTaskResult> createTrigger(CreateTriggerStatement createTriggerStatement);
SettableFuture<ConfigTaskResult> dropTrigger(String triggerName);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
index 27aa838107..acb6bda290 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
@@ -25,8 +25,6 @@ import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.trigger.enums.TriggerEvent;
-import org.apache.iotdb.commons.trigger.enums.TriggerType;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
@@ -41,6 +39,7 @@ import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowTTLTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeSinkTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
@@ -232,20 +231,16 @@ public class StandaloneConfigTaskExecutor implements IConfigTaskExecutor {
@Override
public SettableFuture<ConfigTaskResult> createTrigger(
- String triggerName,
- String className,
- String jarPath,
- boolean usingURI,
- TriggerEvent triggerEvent,
- TriggerType triggerType,
- PartialPath pathPattern) {
+ CreateTriggerStatement createTriggerStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
try {
// todo: implementation
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
} catch (Exception e) {
final String message =
- String.format("Failed to create trigger %s, because %s.", triggerName, e.getMessage());
+ String.format(
+ "Failed to create trigger %s, because %s.",
+ createTriggerStatement.getTriggerName(), e.getMessage());
LOGGER.error(message, e);
future.setException(
new IoTDBException(message, TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateTriggerTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateTriggerTask.java
index 98d3085aea..701199e9e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateTriggerTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateTriggerTask.java
@@ -19,9 +19,6 @@
package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.trigger.enums.TriggerEvent;
-import org.apache.iotdb.commons.trigger.enums.TriggerType;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
@@ -30,34 +27,16 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import com.google.common.util.concurrent.ListenableFuture;
public class CreateTriggerTask implements IConfigTask {
- private final String triggerName;
- private final String className;
-
- private final String jarPath;
-
- private final boolean usingURI;
-
- private final TriggerEvent triggerEvent;
-
- private final TriggerType triggerType;
-
- private final PartialPath pathPattern;
+ private final CreateTriggerStatement createTriggerStatement;
public CreateTriggerTask(CreateTriggerStatement createTriggerStatement) {
- this.triggerName = createTriggerStatement.getTriggerName();
- this.className = createTriggerStatement.getClassName();
- this.jarPath = createTriggerStatement.getJarPath();
- this.usingURI = createTriggerStatement.isUsingURI();
- this.triggerEvent = createTriggerStatement.getTriggerEvent();
- this.triggerType = createTriggerStatement.getTriggerType();
- this.pathPattern = createTriggerStatement.getPathPattern();
+ this.createTriggerStatement = createTriggerStatement;
}
@Override
public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
- return configTaskExecutor.createTrigger(
- triggerName, className, jarPath, usingURI, triggerEvent, triggerType, pathPattern);
+ return configTaskExecutor.createTrigger(createTriggerStatement);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 390bc52045..1282b32e8f 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -31,6 +31,8 @@ import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.service.JMXService;
import org.apache.iotdb.commons.service.RegisterManager;
import org.apache.iotdb.commons.service.StartupChecks;
+import org.apache.iotdb.commons.trigger.service.TriggerClassLoaderManager;
+import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFClassLoaderManager;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
@@ -62,6 +64,7 @@ import org.apache.iotdb.db.service.basic.StandaloneServiceProvider;
import org.apache.iotdb.db.service.metrics.MetricService;
import org.apache.iotdb.db.service.thrift.impl.ClientRPCServiceImpl;
import org.apache.iotdb.db.sync.SyncService;
+import org.apache.iotdb.db.trigger.service.TriggerManagementService;
import org.apache.iotdb.db.wal.WALManager;
import org.apache.iotdb.db.wal.utils.WALMode;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -287,6 +290,7 @@ public class DataNode implements DataNodeMBean {
registerManager.register(DriverScheduler.getInstance());
registerUdfServices();
+ registerTriggerServices();
logger.info(
"IoTDB DataNode is setting up, some storage groups may not be ready now, please wait several seconds...");
@@ -366,7 +370,7 @@ public class DataNode implements DataNodeMBean {
registerManager.register(TemporaryQueryDataFileService.getInstance());
registerManager.register(
UDFExecutableManager.setupAndGetInstance(
- IoTDBDescriptor.getInstance().getConfig().getTemporaryLibDir(),
+ IoTDBDescriptor.getInstance().getConfig().getUdfTemporaryLibDir(),
IoTDBDescriptor.getInstance().getConfig().getUdfDir()));
registerManager.register(
UDFClassLoaderManager.setupAndGetInstance(
@@ -379,6 +383,17 @@ public class DataNode implements DataNodeMBean {
+ File.separator));
}
+ private void registerTriggerServices() throws StartupException {
+ registerManager.register(
+ TriggerExecutableManager.setupAndGetInstance(
+ IoTDBDescriptor.getInstance().getConfig().getTriggerTemporaryLibDir(),
+ IoTDBDescriptor.getInstance().getConfig().getTriggerDir()));
+ registerManager.register(
+ TriggerClassLoaderManager.setupAndGetInstance(
+ IoTDBDescriptor.getInstance().getConfig().getTriggerDir()));
+ registerManager.register(TriggerManagementService.setupAndGetInstance());
+ }
+
private void initSchemaEngine() {
long time = System.currentTimeMillis();
SchemaEngine.getInstance().init();
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 3394d5e2d9..81362ca88f 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -36,8 +36,11 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.trigger.TriggerInformation;
+import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
+import org.apache.iotdb.confignode.rpc.thrift.TTriggerState;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse;
import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
@@ -75,6 +78,7 @@ import org.apache.iotdb.db.service.RegionMigrateService;
import org.apache.iotdb.db.service.metrics.MetricService;
import org.apache.iotdb.db.service.metrics.enums.Metric;
import org.apache.iotdb.db.service.metrics.enums.Tag;
+import org.apache.iotdb.db.trigger.service.TriggerManagementService;
import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
import org.apache.iotdb.metrics.type.Gauge;
import org.apache.iotdb.metrics.utils.MetricLevel;
@@ -107,6 +111,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateConfigNodeGroupReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
+import org.apache.iotdb.mpp.rpc.thrift.TactiveTriggerInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TcreateTriggerInstanceReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -228,7 +233,9 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
}
if (groupId instanceof DataRegionId) {
+
writeResponse = DataRegionConsensusImpl.getInstance().write(groupId, planNode);
+
} else {
writeResponse = SchemaRegionConsensusImpl.getInstance().write(groupId, planNode);
}
@@ -760,7 +767,37 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus createTriggerInstance(TcreateTriggerInstanceReq req) throws TException {
- // todo: implementation
+ TriggerInformation triggerInformation = TriggerInformation.deserialize(req.triggerInformation);
+ try {
+ // set state to INACTIVE when creating trigger instance
+ triggerInformation.setTriggerState(TTriggerState.INACTIVE);
+ // save jar file at trigger_lib_dir
+ TriggerExecutableManager.getInstance()
+ .writeToLibDir(req.jarFile, triggerInformation.getJarName());
+ // register trigger information with TriggerRegistrationService
+ // config nodes take responsibility for synchronization control
+ TriggerManagementService.getInstance().register(triggerInformation);
+ } catch (Exception e) {
+ LOGGER.warn(
+ "Error occurred when creating trigger instance for trigger: {}. The cause is {}.",
+ triggerInformation.getTriggerName(),
+ e.getMessage());
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage());
+ }
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ @Override
+ public TSStatus activeTriggerInstance(TactiveTriggerInstanceReq req) throws TException {
+ try {
+ TriggerManagementService.getInstance().activeTrigger(req.triggerName);
+ } catch (Exception e) {
+ LOGGER.error("Error occurred during ");
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage());
+ }
+
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerExecutor.java b/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerExecutor.java
new file mode 100644
index 0000000000..9227dbe75e
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerExecutor.java
@@ -0,0 +1,57 @@
+/*
+ * 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.iotdb.db.trigger.executor;
+
+import org.apache.iotdb.commons.trigger.TriggerInformation;
+import org.apache.iotdb.commons.trigger.exception.TriggerExecutionException;
+import org.apache.iotdb.trigger.api.Trigger;
+import org.apache.iotdb.trigger.api.TriggerAttributes;
+
+public class TriggerExecutor {
+ private final TriggerInformation triggerInformation;
+
+ private final Trigger trigger;
+
+ public TriggerExecutor(TriggerInformation triggerInformation, Trigger trigger) {
+ this.triggerInformation = triggerInformation;
+ this.trigger = trigger;
+ // call Trigger#validate and Trigger#onCreate
+ onCreate();
+ }
+
+ private void onCreate() {
+ try {
+ TriggerAttributes attributes = new TriggerAttributes(triggerInformation.getAttributes());
+ trigger.validate(attributes);
+ trigger.onCreate(attributes);
+ } catch (Exception e) {
+ onTriggerExecutionError("validate/onCreate(TriggerAttributes)", e);
+ }
+ }
+
+ private void onTriggerExecutionError(String methodName, Exception e)
+ throws TriggerExecutionException {
+ throw new TriggerExecutionException(
+ String.format(
+ "Error occurred during executing Trigger#%s: %s",
+ methodName, System.lineSeparator())
+ + e);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/trigger/service/TriggerManagementService.java b/server/src/main/java/org/apache/iotdb/db/trigger/service/TriggerManagementService.java
new file mode 100644
index 0000000000..6b63006df6
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/trigger/service/TriggerManagementService.java
@@ -0,0 +1,189 @@
+/*
+ * 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.iotdb.db.trigger.service;
+
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.exception.StartupException;
+import org.apache.iotdb.commons.service.IService;
+import org.apache.iotdb.commons.service.ServiceType;
+import org.apache.iotdb.commons.trigger.TriggerInformation;
+import org.apache.iotdb.commons.trigger.TriggerTable;
+import org.apache.iotdb.commons.trigger.exception.TriggerManagementException;
+import org.apache.iotdb.commons.trigger.service.TriggerClassLoader;
+import org.apache.iotdb.commons.trigger.service.TriggerClassLoaderManager;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.trigger.executor.TriggerExecutor;
+import org.apache.iotdb.trigger.api.Trigger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class TriggerManagementService implements IService {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TriggerManagementService.class);
+
+ private final ReentrantLock registrationLock;
+
+ private final TriggerTable triggerTable;
+
+ private final Map<String, TriggerExecutor> executorMap;
+
+ private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
+
+ private TDataNodeLocation tDataNodeLocationCache;
+
+ private TriggerManagementService() {
+ this.registrationLock = new ReentrantLock();
+ this.triggerTable = new TriggerTable();
+ this.executorMap = new ConcurrentHashMap<>();
+ }
+
+ public void acquireRegistrationLock() {
+ registrationLock.lock();
+ }
+
+ public void releaseRegistrationLock() {
+ registrationLock.unlock();
+ }
+
+ public void register(TriggerInformation triggerInformation) {
+ acquireRegistrationLock();
+ checkIfRegistered(triggerInformation);
+ doRegister(triggerInformation);
+ releaseRegistrationLock();
+ };
+
+ public void activeTrigger(String triggerName) {
+ triggerTable.activeTrigger(triggerName);
+ };
+
+ private void checkIfRegistered(TriggerInformation triggerInformation)
+ throws TriggerManagementException {
+ String triggerName = triggerInformation.getTriggerName();
+ if (triggerTable.containsTrigger(triggerName)) {
+ String errorMessage =
+ String.format(
+ "Failed to registered trigger %s, "
+ + "because trigger %s has already been registered in TriggerTable",
+ triggerName, triggerName);
+ LOGGER.warn(errorMessage);
+ throw new TriggerManagementException(errorMessage);
+ }
+ }
+
+ private void doRegister(TriggerInformation triggerInformation) {
+ try (TriggerClassLoader currentActiveClassLoader =
+ TriggerClassLoaderManager.getInstance().updateAndGetActiveClassLoader()) {
+ String triggerName = triggerInformation.getTriggerName();
+ triggerTable.addTriggerInformation(triggerName, triggerInformation);
+ // if it is a stateful trigger, we only create its instance on specified DataNode
+ if (!triggerInformation.isStateful()
+ || triggerInformation.getDataNodeLocation().equals(getTDataNodeLocation())) {
+ // get trigger instance
+ Trigger trigger =
+ constructTriggerInstance(triggerInformation.getClassName(), currentActiveClassLoader);
+ // construct and save TriggerExecutor
+ TriggerExecutor triggerExecutor = new TriggerExecutor(triggerInformation, trigger);
+ executorMap.put(triggerName, triggerExecutor);
+ }
+ } catch (Exception e) {
+ String errorMessage =
+ String.format(
+ "Failed to register trigger %s with className: %s. The cause is: %s",
+ triggerInformation.getTriggerName(),
+ triggerInformation.getClassName(),
+ e.getMessage());
+ LOGGER.warn(errorMessage);
+ throw new TriggerManagementException(errorMessage);
+ }
+ }
+
+ private Trigger constructTriggerInstance(String className, TriggerClassLoader classLoader)
+ throws TriggerManagementException {
+ try {
+ Class<?> triggerClass = Class.forName(className, true, classLoader);
+ return (Trigger) triggerClass.getDeclaredConstructor().newInstance();
+ } catch (InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException e) {
+ throw new TriggerManagementException(
+ String.format(
+ "Failed to reflect trigger instance with className(%s), because %s", className, e));
+ }
+ }
+
+ private TDataNodeLocation getTDataNodeLocation() {
+ if (tDataNodeLocationCache == null) {
+ // Set DataNodeLocation
+ tDataNodeLocationCache = new TDataNodeLocation();
+ tDataNodeLocationCache.setDataNodeId(CONFIG.getDataNodeId());
+ tDataNodeLocationCache.setClientRpcEndPoint(
+ new TEndPoint(CONFIG.getRpcAddress(), CONFIG.getRpcPort()));
+ tDataNodeLocationCache.setInternalEndPoint(
+ new TEndPoint(CONFIG.getInternalAddress(), CONFIG.getInternalPort()));
+ tDataNodeLocationCache.setMPPDataExchangeEndPoint(
+ new TEndPoint(CONFIG.getInternalAddress(), CONFIG.getMppDataExchangePort()));
+ tDataNodeLocationCache.setDataRegionConsensusEndPoint(
+ new TEndPoint(CONFIG.getInternalAddress(), CONFIG.getDataRegionConsensusPort()));
+ tDataNodeLocationCache.setSchemaRegionConsensusEndPoint(
+ new TEndPoint(CONFIG.getInternalAddress(), CONFIG.getSchemaRegionConsensusPort()));
+ }
+ return tDataNodeLocationCache;
+ }
+
+ @Override
+ public void start() throws StartupException {}
+
+ @Override
+ public void stop() {
+ // nothing to do
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.TRIGGER_REGISTRATION_SERVICE;
+ }
+
+ /////////////////////////////////////////////////////////////////////////////////////////////////
+ // singleton instance holder
+ /////////////////////////////////////////////////////////////////////////////////////////////////
+
+ private static TriggerManagementService INSTANCE = null;
+
+ public static synchronized TriggerManagementService setupAndGetInstance() {
+ if (INSTANCE == null) {
+ INSTANCE = new TriggerManagementService();
+ }
+ return INSTANCE;
+ }
+
+ public static TriggerManagementService getInstance() {
+ return INSTANCE;
+ }
+}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift b/thrift-confignode/src/main/thrift/confignode.thrift
index af32d66c63..e4c12899c1 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -249,8 +249,14 @@ enum TTriggerState {
struct TCreateTriggerReq {
1: required string triggerName
- 2: required binary triggerInformation
- 3: required common.TFile jarFile
+ 2: required string className,
+ 3: required string jarPath,
+ 4: required bool usingURI,
+ 5: required byte triggerEvent,
+ 6: required byte triggerType
+ 7: required binary pathPattern,
+ 8: required map<string, string> attributes,
+ 9: optional binary jarFile
}
struct TDropTriggerReq {
diff --git a/thrift/src/main/thrift/datanode.thrift b/thrift/src/main/thrift/datanode.thrift
index e8f463852a..d62800c275 100644
--- a/thrift/src/main/thrift/datanode.thrift
+++ b/thrift/src/main/thrift/datanode.thrift
@@ -174,6 +174,10 @@ struct TcreateTriggerInstanceReq {
2: required binary jarFile
}
+struct TactiveTriggerInstanceReq {
+ 1: required string triggerName
+}
+
struct TDropTriggerInstanceReq {
1: required string triggerName
2: required bool needToDeleteJarFile
@@ -261,11 +265,11 @@ service IDataNodeRPCService {
common.TSStatus invalidatePartitionCache(TInvalidateCacheReq req)
/**
- * Config node will invalidate Schema Info cache.
- *
- * @param bool:isStorageGroup, string:fullPath
- */
- common.TSStatus invalidateSchemaCache(TInvalidateCacheReq req)
+ * Config node will invalidate Schema Info cache.
+ *
+ * @param bool:isStorageGroup, string:fullPath
+ */
+ common.TSStatus invalidateSchemaCache(TInvalidateCacheReq req)
/**
* Config node will delete a data/schema region of a certain storageGroup.
@@ -348,19 +352,26 @@ service IDataNodeRPCService {
**/
common.TSStatus dropFunction(TDropFunctionRequest req)
- /**
- * Config node will create a trigger instance on data node.
- *
- * @param TriggerInformation, jar file.
- **/
+ /**
+ * Config node will create a trigger instance on data node.
+ *
+ * @param TriggerInformation, jar file.
+ **/
common.TSStatus createTriggerInstance(TcreateTriggerInstanceReq req)
/**
- * Config node will drop a trigger on all online config nodes and data nodes.
- *
- * @param trigger name, whether need to delete jar
- **/
- common.TSStatus dropTriggerInstance(TDropTriggerInstanceReq req)
+ * Config node will active a trigger instance on data node.
+ *
+ * @param trigger name.
+ **/
+ common.TSStatus activeTriggerInstance(TactiveTriggerInstanceReq req)
+
+ /**
+ * Config node will drop a trigger on all online config nodes and data nodes.
+ *
+ * @param trigger name, whether need to delete jar
+ **/
+ common.TSStatus dropTriggerInstance(TDropTriggerInstanceReq req)
/**
* Config node will invalidate permission Info cache.