You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/07/19 08:18:41 UTC
[iotdb] branch rel/1.2 updated: [IOTDB-6055] Pipe: Enable auto restart when pipes stopped by ConfigNode because of critical exceptions (#10485) (#10590)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new c52055c7030 [IOTDB-6055] Pipe: Enable auto restart when pipes stopped by ConfigNode because of critical exceptions (#10485) (#10590)
c52055c7030 is described below
commit c52055c7030e3910899fc0cfd910e086a7b80e90
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Jul 19 16:18:35 2023 +0800
[IOTDB-6055] Pipe: Enable auto restart when pipes stopped by ConfigNode because of critical exceptions (#10485) (#10590)
Co-authored-by: Steve Yurong Su <ro...@apache.org>
(cherry picked from commit a02323a4844817860b104f9622013eb04e34e201)
---
.../manager/pipe/runtime/PipeHeartbeatParser.java | 6 +-
.../manager/pipe/runtime/PipeMetaSyncer.java | 53 +++++++++-
.../manager/pipe/task/PipeTaskCoordinator.java | 21 ++--
.../confignode/persistence/pipe/PipeTaskInfo.java | 116 +++++++++++++++++++--
.../impl/pipe/AbstractOperatePipeProcedureV2.java | 6 +-
.../impl/pipe/task/StartPipeProcedureV2.java | 5 +-
.../resources/conf/iotdb-common.properties | 9 ++
.../apache/iotdb/commons/conf/CommonConfig.java | 22 +++-
.../iotdb/commons/conf/CommonDescriptor.java | 9 ++
.../iotdb/commons/pipe/config/PipeConfig.java | 8 ++
.../commons/pipe/task/meta/PipeRuntimeMeta.java | 23 +++-
11 files changed, 252 insertions(+), 26 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
index 93132f511c8..fe39e9b397a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorCriticalExcep
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
import org.apache.iotdb.commons.pipe.task.meta.PipeMeta;
+import org.apache.iotdb.commons.pipe.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
@@ -211,7 +212,9 @@ public class PipeHeartbeatParser {
.getStatus()
.get()
.equals(PipeStatus.STOPPED)) {
- pipeMetaOnConfigNode.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
+ PipeRuntimeMeta runtimeMeta = pipeMetaOnConfigNode.getRuntimeMeta();
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.setIsStoppedByRuntimeException(true);
needWriteConsensusOnConfigNodes.set(true);
needPushPipeMetaToDataNodes.set(true);
@@ -240,6 +243,7 @@ public class PipeHeartbeatParser {
exceptionMap.put(dataNodeId, exception);
}
runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.setIsStoppedByRuntimeException(true);
needWriteConsensusOnConfigNodes.set(true);
needPushPipeMetaToDataNodes.set(true);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
index 25950b8606f..745c7ea299e 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.ProcedureManager;
+import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -33,6 +35,8 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
public class PipeMetaSyncer {
@@ -50,6 +54,11 @@ public class PipeMetaSyncer {
private Future<?> metaSyncFuture;
+ private final AtomicInteger pipeAutoRestartRoundCounter = new AtomicInteger(0);
+
+ private final boolean pipeAutoRestartEnabled =
+ PipeConfig.getInstance().getPipeAutoRestartEnabled();
+
PipeMetaSyncer(ConfigManager configManager) {
this.configManager = configManager;
}
@@ -77,10 +86,28 @@ public class PipeMetaSyncer {
}
}
- private synchronized void sync() {
- final TSStatus status = configManager.getProcedureManager().pipeMetaSync();
+ private void sync() {
+ final ProcedureManager procedureManager = configManager.getProcedureManager();
+
+ boolean somePipesNeedRestarting = false;
+
+ if (pipeAutoRestartEnabled
+ && pipeAutoRestartRoundCounter.incrementAndGet()
+ == PipeConfig.getInstance().getPipeMetaSyncerAutoRestartPipeCheckIntervalRound()) {
+ somePipesNeedRestarting = autoRestartWithLock();
+ pipeAutoRestartRoundCounter.set(0);
+ }
+
+ final TSStatus status = procedureManager.pipeMetaSync();
+
+ if (somePipesNeedRestarting
+ && status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ handleSuccessfulRestartWithLock();
+ procedureManager.pipeHandleMetaChange(true, false);
+ }
+
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- configManager.getProcedureManager().pipeHandleMetaChange(true, true);
+ procedureManager.pipeHandleMetaChange(true, true);
}
}
@@ -91,4 +118,24 @@ public class PipeMetaSyncer {
LOGGER.info("PipeMetaSyncer is stopped successfully.");
}
}
+
+ private boolean autoRestartWithLock() {
+ final AtomicReference<PipeTaskInfo> pipeTaskInfo =
+ configManager.getPipeManager().getPipeTaskCoordinator().lock();
+ try {
+ return pipeTaskInfo.get().autoRestart();
+ } finally {
+ configManager.getPipeManager().getPipeTaskCoordinator().unlock();
+ }
+ }
+
+ private void handleSuccessfulRestartWithLock() {
+ final AtomicReference<PipeTaskInfo> pipeTaskInfo =
+ configManager.getPipeManager().getPipeTaskCoordinator().lock();
+ try {
+ pipeTaskInfo.get().handleSuccessfulRestart();
+ } finally {
+ configManager.getPipeManager().getPipeTaskCoordinator().unlock();
+ }
+ }
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
index 5f0d8f68d17..886a728916b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
@@ -65,6 +65,7 @@ public class PipeTaskCoordinator {
*/
public AtomicReference<PipeTaskInfo> lock() {
pipeTaskCoordinatorLock.lock();
+ LOGGER.info("Pipe task coordinator locked.");
pipeTaskInfoHolder = new AtomicReference<>(pipeTaskInfo);
return pipeTaskInfoHolder;
@@ -84,6 +85,7 @@ public class PipeTaskCoordinator {
try {
pipeTaskCoordinatorLock.unlock();
+ LOGGER.info("Pipe task coordinator unlocked.");
return true;
} catch (IllegalMonitorStateException ignored) {
// This is thrown if unlock() is called without lock() called first.
@@ -92,14 +94,13 @@ public class PipeTaskCoordinator {
}
}
- /* Caller should ensure that the method is called in the lock {@link #lock()}. */
+ /** Caller should ensure that the method is called in the lock {@link #lock()}. */
public TSStatus createPipe(TCreatePipeReq req) {
return configManager.getProcedureManager().createPipe(req);
}
- /* Caller should ensure that the method is called in the lock {@link #lock()}. */
+ /** Caller should ensure that the method is called in the lock {@link #lock()}. */
public TSStatus startPipe(String pipeName) {
- // Whether there are exceptions to clear
final boolean hasException = pipeTaskInfo.hasExceptions(pipeName);
final TSStatus status = configManager.getProcedureManager().startPipe(pipeName);
if (status == RpcUtils.SUCCESS_STATUS && hasException) {
@@ -109,12 +110,20 @@ public class PipeTaskCoordinator {
return status;
}
- /* Caller should ensure that the method is called in the lock {@link #lock()}. */
+ /** Caller should ensure that the method is called in the lock {@link #lock()}. */
public TSStatus stopPipe(String pipeName) {
- return configManager.getProcedureManager().stopPipe(pipeName);
+ final boolean isStoppedByRuntimeException = pipeTaskInfo.isStoppedByRuntimeException(pipeName);
+ final TSStatus status = configManager.getProcedureManager().stopPipe(pipeName);
+ if (status == RpcUtils.SUCCESS_STATUS && isStoppedByRuntimeException) {
+ LOGGER.info(
+ "Pipe {} has stopped successfully manually, stop its auto restart process.", pipeName);
+ pipeTaskInfo.setIsStoppedByRuntimeExceptionToFalse(pipeName);
+ configManager.getProcedureManager().pipeHandleMetaChange(true, true);
+ }
+ return status;
}
- /* Caller should ensure that the method is called in the lock {@link #lock()}. */
+ /** Caller should ensure that the method is called in the lock {@link #lock()}. */
public TSStatus dropPipe(String pipeName) {
final boolean isPipeExistedBeforeDrop = pipeTaskInfo.isPipeExisted(pipeName);
final TSStatus status = configManager.getProcedureManager().dropPipe(pipeName);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index dabe428f70a..28ade7c5f5d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -50,6 +50,8 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -376,7 +378,6 @@ public class PipeTaskInfo implements SnapshotProcessor {
}
final PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
-
final Map<Integer, PipeRuntimeException> exceptionMap =
runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
@@ -397,30 +398,48 @@ public class PipeTaskInfo implements SnapshotProcessor {
return hasException.get();
}
+ public boolean isStoppedByRuntimeException(String pipeName) {
+ acquireReadLock();
+ try {
+ return isStoppedByRuntimeExceptionInternal(pipeName);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
+ private boolean isStoppedByRuntimeExceptionInternal(String pipeName) {
+ return pipeMetaKeeper.containsPipeMeta(pipeName)
+ && pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta().getIsStoppedByRuntimeException();
+ }
+
/**
- * Clear the exceptions of a pipe locally after it starts successfully.
+ * Clear the exceptions of, and set the isAutoStopped flag to false for a pipe locally after it
+ * starts successfully.
*
- * <p>If there are exceptions cleared, the messages will then be updated to all the nodes through
- * {@link PipeHandleMetaChangeProcedure}.
+ * <p>If there are exceptions cleared or flag changed, the messages will then be updated to all
+ * the nodes through {@link PipeHandleMetaChangeProcedure}.
*
- * @param pipeName The name of the pipes to be clear exception
+ * @param pipeName The name of the pipe to be clear exception
*/
- public void clearExceptions(String pipeName) {
+ public void clearExceptionsAndSetIsStoppedByRuntimeExceptionToFalse(String pipeName) {
acquireWriteLock();
try {
- clearExceptionsInternal(pipeName);
+ clearExceptionsAndSetIsStoppedByRuntimeExceptionToFalseInternal(pipeName);
} finally {
releaseWriteLock();
}
}
- private void clearExceptionsInternal(String pipeName) {
+ private void clearExceptionsAndSetIsStoppedByRuntimeExceptionToFalseInternal(String pipeName) {
if (!pipeMetaKeeper.containsPipeMeta(pipeName)) {
return;
}
final PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ // To avoid unnecessary retries, we set the isStoppedByRuntimeException flag to false
+ runtimeMeta.setIsStoppedByRuntimeException(false);
+
runtimeMeta.setExceptionsClearTime(System.currentTimeMillis());
final Map<Integer, PipeRuntimeException> exceptionMap =
@@ -428,6 +447,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
if (!exceptionMap.isEmpty()) {
exceptionMap.clear();
}
+
runtimeMeta
.getConsensusGroupId2TaskMetaMap()
.values()
@@ -439,6 +459,23 @@ public class PipeTaskInfo implements SnapshotProcessor {
});
}
+ public void setIsStoppedByRuntimeExceptionToFalse(String pipeName) {
+ acquireWriteLock();
+ try {
+ setIsStoppedByRuntimeExceptionToFalseInternal(pipeName);
+ } finally {
+ releaseWriteLock();
+ }
+ }
+
+ private void setIsStoppedByRuntimeExceptionToFalseInternal(String pipeName) {
+ if (!pipeMetaKeeper.containsPipeMeta(pipeName)) {
+ return;
+ }
+
+ pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta().setIsStoppedByRuntimeException(false);
+ }
+
/**
* Record the exceptions of all pipes locally if they encountered failure when pushing pipe meta.
*
@@ -481,6 +518,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
// Mark the status of the pipe with exception as stopped
runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.setIsStoppedByRuntimeException(true);
final Map<Integer, PipeRuntimeException> exceptionMap =
runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
@@ -499,6 +537,68 @@ public class PipeTaskInfo implements SnapshotProcessor {
return hasException;
}
+ public boolean autoRestart() {
+ acquireWriteLock();
+ try {
+ return autoRestartInternal();
+ } finally {
+ releaseWriteLock();
+ }
+ }
+
+ /**
+ * Set the statuses of all the pipes stopped automatically because of critical exceptions to
+ * {@link PipeStatus#RUNNING} in order to restart them.
+ *
+ * @return true if there are pipes need restarting
+ */
+ private boolean autoRestartInternal() {
+ final AtomicBoolean needRestart = new AtomicBoolean(false);
+ final List<String> pipeToRestart = new LinkedList<>();
+
+ pipeMetaKeeper
+ .getPipeMetaList()
+ .forEach(
+ pipeMeta -> {
+ if (pipeMeta.getRuntimeMeta().getIsStoppedByRuntimeException()) {
+ pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+
+ needRestart.set(true);
+ pipeToRestart.add(pipeMeta.getStaticMeta().getPipeName());
+ }
+ });
+
+ if (needRestart.get()) {
+ LOGGER.info("PipeMetaSyncer is trying to restart the pipes: {}", pipeToRestart);
+ }
+ return needRestart.get();
+ }
+
+ public void handleSuccessfulRestart() {
+ acquireWriteLock();
+ try {
+ handleSuccessfulRestartInternal();
+ } finally {
+ releaseWriteLock();
+ }
+ }
+
+ /**
+ * Clear the exceptions of, and set the isAutoStopped flag to false for the successfully restarted
+ * pipe.
+ */
+ private void handleSuccessfulRestartInternal() {
+ pipeMetaKeeper
+ .getPipeMetaList()
+ .forEach(
+ pipeMeta -> {
+ if (pipeMeta.getRuntimeMeta().getStatus().get().equals(PipeStatus.RUNNING)) {
+ clearExceptionsAndSetIsStoppedByRuntimeExceptionToFalse(
+ pipeMeta.getStaticMeta().getPipeName());
+ }
+ });
+ }
+
/////////////////////////////// Snapshot ///////////////////////////////
@Override
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
index 02f62993fbf..c0ef6d4bfc4 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
@@ -80,11 +80,11 @@ public abstract class AbstractOperatePipeProcedureV2
.getPipeManager()
.getPipeTaskCoordinator()
.lock();
- LOGGER.info("procedureId {} acquire lock.", getProcId());
+ LOGGER.info("ProcedureId {} acquire lock.", getProcId());
return ProcedureLockState.LOCK_ACQUIRED;
}
configNodeProcedureEnv.getNodeLock().waitProcedure(this);
- LOGGER.info("procedureId {} wait for lock.", getProcId());
+ LOGGER.info("ProcedureId {} wait for lock.", getProcId());
return ProcedureLockState.LOCK_EVENT_WAIT;
} finally {
configNodeProcedureEnv.getSchedulerLock().unlock();
@@ -95,7 +95,7 @@ public abstract class AbstractOperatePipeProcedureV2
protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
configNodeProcedureEnv.getSchedulerLock().lock();
try {
- LOGGER.info("procedureId {} release lock.", getProcId());
+ LOGGER.info("ProcedureId {} release lock.", getProcId());
if (pipeTaskInfo != null) {
configNodeProcedureEnv
.getConfigManager()
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
index fa21c6b816d..eae446df24b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
@@ -95,8 +95,9 @@ public class StartPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
String.format("Failed to start pipe %s, details: %s", pipeName, exceptionMessage));
}
- // Clear exception messages if succeeded
- pipeTaskInfo.get().clearExceptions(pipeName);
+ // Clear exceptions and set isStoppedByRuntimeException to false if the pipe is
+ // started successfully on all data nodes
+ pipeTaskInfo.get().clearExceptionsAndSetIsStoppedByRuntimeExceptionToFalse(pipeName);
}
@Override
diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
index 46d5aa0a342..73178ed0bf5 100644
--- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
+++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
@@ -997,6 +997,15 @@ cluster_name=defaultCluster
# The sync regular interval (in minutes) for the PipeMetaSyncer service.
# pipe_meta_syncer_sync_interval_minutes=3
+# When the pipe meta synchronization round is a multiple of it, it will restart pipes stopped automatically by critical exceptions.
+# Useless if the pipe_exception_stopped_auto_restart_enabled is set to false.
+# Eg: if pipe_meta_syncer_sync_interval_minutes=3, and pipe_meta_syncer_auto_restart_pipe_round=5,
+# then the pipe restart process's interval will be 15 minutes, and is executed by the pipe meta sync thread.
+# pipe_meta_syncer_auto_restart_pipe_check_interval_round=1
+
+# Whether to enable pipe auto restart after pipe meets error during runtime.
+# pipe_auto_restart_enabled=true
+
####################
### RatisConsensus Configuration
####################
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 738cd723b59..f41362dfdcf 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -106,7 +106,7 @@ public class CommonConfig {
*/
private int selectorNumOfClientManager = 1;
- /** whether to use thrift compression. */
+ /** Whether to use thrift compression. */
private boolean isRpcThriftCompressionEnabled = false;
private int coreClientNumForEachNode = DefaultProperty.CORE_CLIENT_NUM_FOR_EACH_NODE;
@@ -164,6 +164,8 @@ public class CommonConfig {
private int pipeHeartbeatIntervalSecondsForCollectingPipeMeta = 100;
private long pipeMetaSyncerInitialSyncDelayMinutes = 3;
private long pipeMetaSyncerSyncIntervalMinutes = 3;
+ private long pipeMetaSyncerAutoRestartPipeCheckIntervalRound = 1;
+ private boolean pipeAutoRestartEnabled = true;
/** Whether to use persistent schema mode. */
private String schemaEngineMode = "Memory";
@@ -544,6 +546,24 @@ public class CommonConfig {
this.pipeMetaSyncerSyncIntervalMinutes = pipeMetaSyncerSyncIntervalMinutes;
}
+ public long getPipeMetaSyncerAutoRestartPipeCheckIntervalRound() {
+ return pipeMetaSyncerAutoRestartPipeCheckIntervalRound;
+ }
+
+ public void setPipeMetaSyncerAutoRestartPipeCheckIntervalRound(
+ long pipeMetaSyncerAutoRestartPipeCheckIntervalRound) {
+ this.pipeMetaSyncerAutoRestartPipeCheckIntervalRound =
+ pipeMetaSyncerAutoRestartPipeCheckIntervalRound;
+ }
+
+ public boolean getPipeAutoRestartEnabled() {
+ return pipeAutoRestartEnabled;
+ }
+
+ public void setPipeAutoRestartEnabled(boolean pipeAutoRestartEnabled) {
+ this.pipeAutoRestartEnabled = pipeAutoRestartEnabled;
+ }
+
public long getPipeConnectorRetryIntervalMs() {
return pipeConnectorRetryIntervalMs;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index 7924a5edbfa..aaae5ed2a68 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -329,6 +329,15 @@ public class CommonDescriptor {
properties.getProperty(
"pipe_meta_syncer_sync_interval_minutes",
String.valueOf(config.getPipeMetaSyncerSyncIntervalMinutes()))));
+ config.setPipeMetaSyncerAutoRestartPipeCheckIntervalRound(
+ Long.parseLong(
+ properties.getProperty(
+ "pipe_meta_syncer_auto_restart_pipe_check_interval_round",
+ String.valueOf(config.getPipeMetaSyncerAutoRestartPipeCheckIntervalRound()))));
+ config.setPipeAutoRestartEnabled(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "pipe_auto_restart_enabled", String.valueOf(config.getPipeAutoRestartEnabled()))));
}
public void loadGlobalConfig(TGlobalConfig globalConfig) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index 1dda5f3e347..73bcfd9b19b 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -109,6 +109,14 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeMetaSyncerSyncIntervalMinutes();
}
+ public long getPipeMetaSyncerAutoRestartPipeCheckIntervalRound() {
+ return COMMON_CONFIG.getPipeMetaSyncerAutoRestartPipeCheckIntervalRound();
+ }
+
+ public boolean getPipeAutoRestartEnabled() {
+ return COMMON_CONFIG.getPipeAutoRestartEnabled();
+ }
+
/////////////////////////////// Utils ///////////////////////////////
private static final Logger LOGGER = LoggerFactory.getLogger(PipeConfig.class);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
index 6de10a08edd..a773407d979 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
@@ -37,6 +37,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -62,6 +63,8 @@ public class PipeRuntimeMeta {
private final AtomicLong exceptionsClearTime = new AtomicLong(Long.MIN_VALUE);
+ private final AtomicBoolean isStoppedByRuntimeException = new AtomicBoolean(false);
+
public PipeRuntimeMeta() {
consensusGroupId2TaskMetaMap = new ConcurrentHashMap<>();
}
@@ -92,6 +95,14 @@ public class PipeRuntimeMeta {
}
}
+ public boolean getIsStoppedByRuntimeException() {
+ return isStoppedByRuntimeException.get();
+ }
+
+ public void setIsStoppedByRuntimeException(boolean isStoppedByRuntimeException) {
+ this.isStoppedByRuntimeException.set(isStoppedByRuntimeException);
+ }
+
public ByteBuffer serialize() throws IOException {
PublicBAOS byteArrayOutputStream = new PublicBAOS();
DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
@@ -125,6 +136,7 @@ public class PipeRuntimeMeta {
}
ReadWriteIOUtils.write(exceptionsClearTime.get(), outputStream);
+ ReadWriteIOUtils.write(isStoppedByRuntimeException.get(), outputStream);
}
public void serialize(FileOutputStream outputStream) throws IOException {
@@ -153,6 +165,7 @@ public class PipeRuntimeMeta {
}
ReadWriteIOUtils.write(exceptionsClearTime.get(), outputStream);
+ ReadWriteIOUtils.write(isStoppedByRuntimeException.get(), outputStream);
}
public static PipeRuntimeMeta deserialize(InputStream inputStream) throws IOException {
@@ -208,6 +221,7 @@ public class PipeRuntimeMeta {
}
pipeRuntimeMeta.exceptionsClearTime.set(ReadWriteIOUtils.readLong(inputStream));
+ pipeRuntimeMeta.isStoppedByRuntimeException.set(ReadWriteIOUtils.readBool(inputStream));
return pipeRuntimeMeta;
}
@@ -265,6 +279,7 @@ public class PipeRuntimeMeta {
}
pipeRuntimeMeta.exceptionsClearTime.set(ReadWriteIOUtils.readLong(byteBuffer));
+ pipeRuntimeMeta.isStoppedByRuntimeException.set(ReadWriteIOUtils.readBool(byteBuffer));
return pipeRuntimeMeta;
}
@@ -281,7 +296,8 @@ public class PipeRuntimeMeta {
return Objects.equals(status.get().getType(), that.status.get().getType())
&& consensusGroupId2TaskMetaMap.equals(that.consensusGroupId2TaskMetaMap)
&& dataNodeId2PipeRuntimeExceptionMap.equals(that.dataNodeId2PipeRuntimeExceptionMap)
- && exceptionsClearTime.get() == that.exceptionsClearTime.get();
+ && exceptionsClearTime.get() == that.exceptionsClearTime.get()
+ && isStoppedByRuntimeException.get() == that.isStoppedByRuntimeException.get();
}
@Override
@@ -290,7 +306,8 @@ public class PipeRuntimeMeta {
status,
consensusGroupId2TaskMetaMap,
dataNodeId2PipeRuntimeExceptionMap,
- exceptionsClearTime.get());
+ exceptionsClearTime.get(),
+ isStoppedByRuntimeException.get());
}
@Override
@@ -304,6 +321,8 @@ public class PipeRuntimeMeta {
+ dataNodeId2PipeRuntimeExceptionMap
+ ", exceptionsClearTime="
+ exceptionsClearTime.get()
+ + ", isStoppedByRuntimeException="
+ + isStoppedByRuntimeException.get()
+ "}";
}
}