You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Caideyipi (via GitHub)" <gi...@apache.org> on 2023/07/04 09:27:48 UTC
[GitHub] [iotdb] Caideyipi opened a new pull request, #10433: Pipe: Upgraded the exception handling logic
Caideyipi opened a new pull request, #10433:
URL: https://github.com/apache/iotdb/pull/10433
## Description
Upgraded the exception handle logic, including:
### Time stamp record
Added time stamp for each PipeException, representing the time that the exception is encountered.
### PipeMeta exception tracking
Handled the exceptions thrown after pushing pipe meta, by stopping the related pipe and recording the exceptions.
### Pipe create/start/stop/drop procedure handle
Now these procedures will only fail when the operated pipe meets error when pushing pipe meta.
### Exception clear after starting pipe
Clear the exceptions after the pipe restarted successfully, and all the previous exceptions reported are ignored.
### Show pipe upgrade
Added the pipe meta exceptions encountered for each pipe, and added time stamp display for each exception.
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error
conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition,
design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design
(or naming) decision point and compare the alternatives with the designs that you've implemented
(or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere
(e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list),
link to that discussion from this PR description and explain what have changed in your final design
compared to your original proposal or the consensus version in the end of the discussion.
If something hasn't changed since the original discussion, you can omit a detailed discussion of
those aspects of the design here, perhaps apart from brief mentioning for the sake of readability
of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
<hr>
This PR has:
- [ ] been self-reviewed.
- [ ] concurrent read
- [ ] concurrent write
- [ ] concurrent read and write
- [ ] added documentation for new or modified features or behaviors.
- [ ] added Javadocs for most classes and all non-trivial methods.
- [ ] added or updated version, __license__, or notice information
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious
for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold
for code coverage.
- [ ] added integration tests.
- [ ] been tested in a test IoTDB cluster.
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items
apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items
from the checklist above are strictly necessary, but it would be very helpful if you at least
self-review the PR. -->
<hr>
##### Key changed/added classes (or packages if there are too many classes) in this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253998174
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,14 +215,68 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+
+ if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
+ AtomicBoolean hasException = new AtomicBoolean(false);
Review Comment:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1254014691
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,14 +215,68 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+
+ if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
+ if (!resp.isSetExceptionMessages()) {
+ exceptionMessageBuilder.append(
+ String.format(
+ "DataNodeId: %s, Message: Internal error while processing pushPipeMeta on dataNodes.",
+ dataNodeId));
+ continue;
+ }
+
+ AtomicBoolean hasException = new AtomicBoolean(false);
+
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ // Ignore the timeStamp for simplicity
+ if (pipeName == null) {
+ hasException.set(true);
+ exceptionMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s ",
+ message.getPipeName(), message.getMessage()));
+ } else if (pipeName.equals(message.getPipeName())) {
+ hasException.set(true);
+ exceptionMessageBuilder.append(
+ String.format("Message: %s ", message.getMessage()));
+ }
+ });
+
+ if (hasException.get()) {
+ // Only print dataNodeId if the given pipe meets exception on that node
+ exceptionMessageBuilder.insert(0, String.format("DataNodeId: %s ", dataNodeId));
Review Comment:
```suggestion
exceptionMessageBuilder.insert(0, String.format("DataNodeId: %s, ", dataNodeId));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253897943
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,101 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ public boolean hasExceptions(String pipeName) {
+ if (!pipeMetaKeeper.containsPipeMeta(pipeName)) {
+ return false;
+ }
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ return true;
+ }
+ AtomicBoolean hasException = new AtomicBoolean(false);
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ hasException.set(true);
+ }
+ });
+ return hasException.get();
+ }
+
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ */
+ public void clearExceptions(String pipeName) {
+ if (!pipeMetaKeeper.containsPipeMeta(pipeName)) {
+ return;
+ }
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ runtimeMeta.setExceptionsClearTime(System.currentTimeMillis());
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ exceptionMap.clear();
+ }
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ pipeTaskMeta.clearExceptionMessages();
+ }
+ });
+ }
+
+ /**
+ * Record the exceptions of all pipes locally if they encountered failure when pushing pipe meta.
+ * If there are exceptions recorded, the related pipes will be stopped, and the exception messages
+ * will then be updated to all the nodes through PipeHandleMetaChangeProcedure.
+ *
+ * @param respMap The responseMap after pushing pipe meta
+ * @return true if there are exceptions encountered
+ */
+ public boolean recordPushPipeMetaExceptions(Map<Integer, TPushPipeMetaResp> respMap) {
+ boolean hasException = false;
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
+ hasException = true;
+ if (!resp.isSetExceptionMessages()) {
Review Comment:
Is this case handled by parsePushPipeMetaExceptionForPipe?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253933629
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java:
##########
@@ -128,21 +191,31 @@ public boolean equals(Object o) {
}
PipeRuntimeMeta that = (PipeRuntimeMeta) o;
return Objects.equals(status.get().getType(), that.status.get().getType())
- && consensusGroupIdToTaskMetaMap.equals(that.consensusGroupIdToTaskMetaMap);
+ && consensusGroupId2TaskMetaMap.equals(that.consensusGroupId2TaskMetaMap)
+ && dataNodeId2PipeRuntimeExceptionMap.equals(that.dataNodeId2PipeRuntimeExceptionMap)
+ && exceptionsClearTime.get() == that.exceptionsClearTime.get();
}
@Override
public int hashCode() {
- return Objects.hash(status, consensusGroupIdToTaskMetaMap);
+ return Objects.hash(
+ status,
+ consensusGroupId2TaskMetaMap,
+ dataNodeId2PipeRuntimeExceptionMap,
+ exceptionsClearTime.get());
}
@Override
public String toString() {
return "PipeRuntimeMeta{"
+ "status="
+ status
- + ", consensusGroupIdToTaskMetaMap="
- + consensusGroupIdToTaskMetaMap
- + '}';
+ + ", consensusGroupId2TaskMetaMap="
+ + consensusGroupId2TaskMetaMap
+ + ", dataNodeId2PipeMetaExceptionMap="
+ + dataNodeId2PipeRuntimeExceptionMap
+ + ", exceptionsClearTime="
+ + exceptionsClearTime
Review Comment:
```suggestion
+ exceptionsClearTime.get()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252549984
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,11 +201,46 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()) {
+ exceptionMessageBuilder.append(String.format("DataNodeId: %s ", dataNodeId));
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ // Ignore the timeStamp for simplicity
+ if (pipeName == null) {
+ exceptionMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s ",
+ message.getPipeName(), message.getMessage()));
+ } else if (pipeName.equals(message.getPipeName())) {
+ exceptionMessageBuilder.append(
+ String.format("Message: %s ", message.getMessage()));
+ }
+ });
+ }
Review Comment:
```suggestion
});
exceptionMessageBuilder.append('. ');
}
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,11 +201,46 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()) {
+ exceptionMessageBuilder.append(String.format("DataNodeId: %s ", dataNodeId));
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ // Ignore the timeStamp for simplicity
+ if (pipeName == null) {
+ exceptionMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s ",
+ message.getPipeName(), message.getMessage()));
+ } else if (pipeName.equals(message.getPipeName())) {
+ exceptionMessageBuilder.append(
+ String.format("Message: %s ", message.getMessage()));
+ }
+ });
+ }
Review Comment:
```suggestion
});
exceptionMessageBuilder.append(". ");
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252547971
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java:
##########
@@ -48,56 +51,66 @@ protected PipeTaskOperation getOperation() {
protected void executeFromValidateTask(ConfigNodeProcedureEnv env) {
LOGGER.info("PipeMetaSyncProcedure: executeFromValidateTask");
- // do nothing
+ // Do nothing
}
@Override
protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
LOGGER.info("PipeMetaSyncProcedure: executeFromCalculateInfoForTask");
- // do nothing
+ // Do nothing
}
@Override
protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
LOGGER.info("PipeMetaSyncProcedure: executeFromWriteConfigNodeConsensus");
- // do nothing
+ // Do nothing
}
@Override
- protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
+ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
LOGGER.info("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
- pushPipeMetaToDataNodesIgnoreException(env);
+ Map<Integer, TPushPipeMetaResp> respMap = pushPipeMetaToDataNodes(env);
+ if (env.getConfigManager()
+ .getPipeManager()
+ .getPipeTaskCoordinator()
+ .getPipeTaskInfo()
+ .recordPushPipeMetaExceptions(respMap)) {
+ throw new PipeException(
+ String.format(
+ "Failed to push pipe meta to dataNodes, details: %s",
+ parsePushPipeMetaExceptionForPipe(null, respMap)));
Review Comment:
use logger instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu merged pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu merged PR #10433:
URL: https://github.com/apache/iotdb/pull/10433
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252537507
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,71 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ * @return true if there are exceptions cleared
+ */
+ public boolean clearExceptions(String pipeName) {
+ AtomicBoolean isCleared = new AtomicBoolean(false);
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
Review Comment:
getPipeMeta may cause NPE
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,71 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ * @return true if there are exceptions cleared
+ */
+ public boolean clearExceptions(String pipeName) {
+ AtomicBoolean isCleared = new AtomicBoolean(false);
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
Review Comment:
check it before use
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252530417
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java:
##########
@@ -66,7 +66,12 @@ public TSStatus createPipe(TCreatePipeReq req) {
}
public TSStatus startPipe(String pipeName) {
- return configManager.getProcedureManager().startPipe(pipeName);
+ TSStatus status = configManager.getProcedureManager().startPipe(pipeName);
+ if (status == RpcUtils.SUCCESS_STATUS && pipeTaskInfo.clearExceptions(pipeName)) {
+ LOGGER.info("Pipe {} has started successfully, clear its exceptions.", pipeName);
+ configManager.getProcedureManager().pipeHandleMetaChange(true, true);
+ }
Review Comment:
Move it to StartPipeProcedureV2#executeFromOperateOnDataNodes to avoid concurrent modification caused by clearExceptions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253933798
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java:
##########
@@ -32,29 +34,54 @@
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public class PipeRuntimeMeta {
private final AtomicReference<PipeStatus> status;
Review Comment:
```suggestion
private final AtomicReference<PipeStatus> status = new ...;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252554315
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java:
##########
@@ -86,16 +85,11 @@ protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
}
@Override
- protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeException, IOException {
+ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("DropPipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName);
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- throw new PipeException(
- String.format(
- "Failed to drop pipe %s on data nodes. Failures: %s", pipeName, result.getMessage()));
- }
+ // Will not rollback
+ pushPipeMetaToDataNodesIgnoreException(env);
Review Comment:
Why make a change here? The PipeException thrown here is used to trigger retries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253897821
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,101 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ public boolean hasExceptions(String pipeName) {
+ if (!pipeMetaKeeper.containsPipeMeta(pipeName)) {
+ return false;
+ }
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ return true;
+ }
+ AtomicBoolean hasException = new AtomicBoolean(false);
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ hasException.set(true);
+ }
+ });
+ return hasException.get();
+ }
+
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ */
+ public void clearExceptions(String pipeName) {
+ if (!pipeMetaKeeper.containsPipeMeta(pipeName)) {
+ return;
+ }
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ runtimeMeta.setExceptionsClearTime(System.currentTimeMillis());
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ exceptionMap.clear();
+ }
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ pipeTaskMeta.clearExceptionMessages();
+ }
+ });
+ }
+
+ /**
+ * Record the exceptions of all pipes locally if they encountered failure when pushing pipe meta.
+ * If there are exceptions recorded, the related pipes will be stopped, and the exception messages
+ * will then be updated to all the nodes through PipeHandleMetaChangeProcedure.
+ *
+ * @param respMap The responseMap after pushing pipe meta
+ * @return true if there are exceptions encountered
+ */
+ public boolean recordPushPipeMetaExceptions(Map<Integer, TPushPipeMetaResp> respMap) {
+ boolean hasException = false;
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
+ hasException = true;
+ if (!resp.isSetExceptionMessages()) {
Review Comment:
What if hasException is true and resp.isSetExceptionMessages() is false? What message will be sent back to user?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252509915
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java:
##########
@@ -929,18 +931,21 @@ public TSStatus alterView(TAlterViewReq req) {
}
@Override
- public TSStatus pushPipeMeta(TPushPipeMetaReq req) {
+ public TPushPipeMetaResp pushPipeMeta(TPushPipeMetaReq req) {
final List<PipeMeta> pipeMetas = new ArrayList<>();
for (ByteBuffer byteBuffer : req.getPipeMetas()) {
pipeMetas.add(PipeMeta.deserialize(byteBuffer));
}
- try {
- PipeAgent.task().handlePipeMetaChanges(pipeMetas);
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- } catch (Exception e) {
- LOGGER.error("Error occurred when pushing pipe meta", e);
- return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
+ TPushPipeMetaResp resp = new TPushPipeMetaResp();
+ List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
+ PipeAgent.task().handlePipeMetaChanges(pipeMetas);
+ if (exceptionMessages.isEmpty()) {
+ resp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ } else {
+ resp.setStatus(new TSStatus(TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()));
+ resp.setExceptionMessages(exceptionMessages);
}
+ return resp;
Review Comment:
```suggestion
final List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
PipeAgent.task().handlePipeMetaChanges(pipeMetas);
return exceptionMessages.isEmpty()
? new TPushPipeMetaResp()
.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))
: new TPushPipeMetaResp()
.setStatus(new TSStatus(TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()))
.setExceptionMessages(exceptionMessages);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252500684
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/PushPipeMetaRPCHandler.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.confignode.client.async.handlers.rpc;
+
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.confignode.client.DataNodeRequestType;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+public class PushPipeMetaRPCHandler extends AbstractAsyncRPCHandler<TPushPipeMetaResp> {
Review Comment:
PipePushMetaRPCHandler
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252553650
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java:
##########
@@ -190,12 +190,14 @@ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws
"CreatePipeProcedureV2: rollbackFromOperateOnDataNodes({})",
createPipeRequest.getPipeName());
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(
+ createPipeRequest.getPipeName(), pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to rollback create pipe %s on data nodes. Failures: %s",
- pipeStaticMeta.getPipeName(), result.getMessage()));
+ "Failed to rollback create pipe for pipe %s, details: %s",
Review Comment:
```suggestion
"Failed to rollback create pipe %s, details: %s",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] Caideyipi commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253945986
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/AliasIndexPage.java:
##########
@@ -274,21 +278,20 @@ public String splitByKey(String key, String entry, ByteBuffer dstBuffer, boolean
int sp; // virtual index to split
if (monotonic) {
- // new entry into part with more space
- sp = key.compareTo(lastKey) > 0 ? Math.max(pos, n / 2) : Math.min(pos, n / 2);
+ // New entry into part with more space
+ sp = key.compareTo(lastKey) > 0 ? Math.max(pos, n >> 1) : Math.min(pos, n >> 1);
} else {
- sp = n / 2;
+ sp = n >> 1;
}
- // little different from InternalSegment, only the front edge key can not split
+ // Little different from InternalSegment, only the front edge key can not split
sp = sp <= 0 ? 1 : sp;
- // this method BREAKS envelop of the passing in buffer to be more efficient
+ // This method BREAKS envelop of the passing in buffer to be more efficient
// attributes for dstBuffer
- short memberNum = 0,
- spareOffset = (short) dstBuffer.capacity(),
- spareSize = (short) (spareOffset - SchemaFileConfig.PAGE_HEADER_SIZE);
- long nextSegAddress = this.nextPage;
Review Comment:
This variable is not used at all...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253934244
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java:
##########
@@ -198,7 +205,8 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
.getStatus()
.get()
.equals(PipeStatus.STOPPED)) {
- pipeMetaOnConfigNode.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
+ PipeRuntimeMeta runtimeMeta = pipeMetaOnConfigNode.getRuntimeMeta();
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
Review Comment:
Why seperate it ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252498070
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java:
##########
@@ -218,6 +218,7 @@ public enum TSStatusCode {
PIPE_TRANSFER_FILE_ERROR(1804),
PIPE_TRANSFER_EXECUTE_STATEMENT_ERROR(1805),
PIPE_NOT_EXIST_ERROR(1806),
+ PUSH_PIPE_META_ERROR(1807),
Review Comment:
```suggestion
PIPE_PUSH_META_ERROR(1807),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252554750
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java:
##########
@@ -93,11 +92,12 @@ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
throws PipeException, IOException {
LOGGER.info("StopPipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName);
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(pipeName, pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to stop pipe %s on data nodes. Failures: %s", pipeName, result.getMessage()));
+ "Failed to stop pipe for pipe %s, details: %s", pipeName, exceptionMessage));
Review Comment:
```suggestion
"Failed to stop pipe %s, details: %s", pipeName, exceptionMessage));
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java:
##########
@@ -131,12 +131,12 @@ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
throws PipeException, IOException {
LOGGER.info("StopPipeProcedureV2: rollbackFromOperateOnDataNodes({})", pipeName);
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(pipeName, pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to rollback stop pipe %s on data nodes. Failures: %s",
- pipeName, result.getMessage()));
+ "Failed to rollback stop pipe for pipe %s, details: %s", pipeName, exceptionMessage));
Review Comment:
```suggestion
"Failed to rollback stop pipe %s, details: %s", pipeName, exceptionMessage));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252538268
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,71 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ * @return true if there are exceptions cleared
+ */
+ public boolean clearExceptions(String pipeName) {
+ AtomicBoolean isCleared = new AtomicBoolean(false);
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ runtimeMeta.setExceptionsClearTime(System.currentTimeMillis());
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ exceptionMap.clear();
+ isCleared.set(true);
+ }
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ isCleared.set(true);
+ pipeTaskMeta.clearExceptionMessages();
+ }
+ });
+ return isCleared.get();
+ }
+
+ /**
+ * Record the exceptions of all pipes locally if they encountered failure when pushing pipe meta.
+ * If there are exceptions recorded, the related pipes will be stopped, and the exception messages
+ * will then be updated to all the nodes through PipeHandleMetaChangeProcedure.
+ *
+ * @param respMap The responseMap after pushing pipe meta
+ * @return true if there are exceptions encountered
+ */
+ public boolean recordPushPipeMetaExceptions(Map<Integer, TPushPipeMetaResp> respMap) {
+ boolean hasException = false;
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()) {
+ hasException = true;
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ PipeRuntimeMeta runtimeMeta =
+ pipeMetaKeeper.getPipeMeta(message.getPipeName()).getRuntimeMeta();
Review Comment:
getPipeMeta may cause NPE
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252530821
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,71 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ * @return true if there are exceptions cleared
+ */
+ public boolean clearExceptions(String pipeName) {
+ AtomicBoolean isCleared = new AtomicBoolean(false);
Review Comment:
boolean is enough
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253889390
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java:
##########
@@ -131,12 +130,12 @@ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
throws PipeException, IOException {
LOGGER.info("StopPipeProcedureV2: rollbackFromOperateOnDataNodes({})", pipeName);
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(pipeName, pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to rollback stop pipe %s on data nodes. Failures: %s",
- pipeName, result.getMessage()));
+ "Failed to rollback stop pipe %s, details: %s", pipeName, exceptionMessage));
Review Comment:
Should we clear exceptions here? I think not cleaning is ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253991732
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java:
##########
@@ -66,7 +66,19 @@ public TSStatus createPipe(TCreatePipeReq req) {
}
public TSStatus startPipe(String pipeName) {
- return configManager.getProcedureManager().startPipe(pipeName);
+ // Whether there are exceptions to clear
+ lock();
+ try {
+ boolean hasException = pipeTaskInfo.hasExceptions(pipeName);
Review Comment:
There must be a compiler error
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252515535
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -80,10 +80,7 @@ public synchronized void start() {
private synchronized void sync() {
final TSStatus status = configManager.getProcedureManager().pipeMetaSync();
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(
- "PipeMetaSyncer meets error in syncing pipe meta, code: {}, message: {}",
- status.getCode(),
- status.getMessage());
+ configManager.getProcedureManager().pipeHandleMetaChange(true, true);
Review Comment:
Keep the logger here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252506246
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java:
##########
@@ -41,6 +41,13 @@ public class PipeTaskMeta {
private final AtomicReference<ProgressIndex> progressIndex = new AtomicReference<>();
private final AtomicInteger leaderDataNodeId = new AtomicInteger(0);
+
+ /**
+ * Stores the exceptions encountered during run time of each pipe task. The exceptions are
+ * instances of PipeRuntimeCriticalException, PipeRuntimeConnectorCriticalException and
+ * PipeRuntimeNonCriticalException. The failure of them, respectively, will lead to the stop of
Review Comment:
Only CriticalException will lead to the stop of the pipe.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252551844
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,11 +201,46 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()) {
+ exceptionMessageBuilder.append(String.format("DataNodeId: %s ", dataNodeId));
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ // Ignore the timeStamp for simplicity
+ if (pipeName == null) {
+ exceptionMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s ",
Review Comment:
```suggestion
"PipeName: %s, Message: %s; ",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252522532
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java:
##########
@@ -32,29 +34,54 @@
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public class PipeRuntimeMeta {
private final AtomicReference<PipeStatus> status;
- private final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupIdToTaskMetaMap;
+ private final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupId2TaskMetaMap;
+ /**
+ * Stores the exceptions encountered during pushing pipeMeta to DataNodes. The exceptions are all
+ * instances of PipeRuntimeCriticalException, so that the failure of pushing pipeMeta will result
+ * in the halt of transferring data.
+ */
+ private final Map<Integer, PipeRuntimeException> dataNodeId2PipeRuntimeExceptionMap;
+
+ private final AtomicLong exceptionsClearTime = new AtomicLong(Long.MIN_VALUE);
Review Comment:
init it in constructors, like the others'
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252519543
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java:
##########
@@ -32,29 +34,54 @@
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public class PipeRuntimeMeta {
private final AtomicReference<PipeStatus> status;
- private final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupIdToTaskMetaMap;
+ private final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupId2TaskMetaMap;
+ /**
+ * Stores the exceptions encountered during pushing pipeMeta to DataNodes. The exceptions are all
+ * instances of PipeRuntimeCriticalException, so that the failure of pushing pipeMeta will result
+ * in the halt of transferring data.
Review Comment:
```suggestion
* in the halt of pipe execution.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252521899
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -99,6 +97,8 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState
executeFromOperateOnDataNodes(env);
env.getConfigManager().getPipeManager().getPipeTaskCoordinator().unlock();
return Flow.NO_MORE_STATE;
+ default:
+ break;
Review Comment:
```suggestion
throw ..;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] Caideyipi commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253944826
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/handler/AuthorizationHandler.java:
##########
@@ -17,35 +17,24 @@
package org.apache.iotdb.db.protocol.rest.handler;
-import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.rpc.TSStatusCode;
import javax.ws.rs.core.Response;
-import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.SecurityContext;
public class AuthorizationHandler {
public Response checkAuthority(SecurityContext securityContext, Statement statement) {
String userName = securityContext.getUserPrincipal().getName();
- try {
- if (!AuthorityChecker.checkAuthorization(statement, userName)) {
- return Response.ok()
- .entity(
- new ExecutionStatus()
- .code(TSStatusCode.NO_PERMISSION.getStatusCode())
- .message(TSStatusCode.NO_PERMISSION.name()))
- .build();
- }
- } catch (AuthException e) {
+ if (!AuthorityChecker.checkAuthorization(statement, userName)) {
Review Comment:
I changed it because the function "checkAuthorization" actually throws no exceptions here...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252508804
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java:
##########
@@ -929,18 +931,21 @@ public TSStatus alterView(TAlterViewReq req) {
}
@Override
- public TSStatus pushPipeMeta(TPushPipeMetaReq req) {
+ public TPushPipeMetaResp pushPipeMeta(TPushPipeMetaReq req) {
final List<PipeMeta> pipeMetas = new ArrayList<>();
for (ByteBuffer byteBuffer : req.getPipeMetas()) {
pipeMetas.add(PipeMeta.deserialize(byteBuffer));
}
- try {
- PipeAgent.task().handlePipeMetaChanges(pipeMetas);
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- } catch (Exception e) {
- LOGGER.error("Error occurred when pushing pipe meta", e);
- return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
+ TPushPipeMetaResp resp = new TPushPipeMetaResp();
Review Comment:
Keep the try catch to handle unexpected runtime exceptions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252511839
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java:
##########
@@ -105,9 +107,20 @@ public TShowPipeResp convertToTShowPipeResp() {
final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
final StringBuilder exceptionMessageBuilder = new StringBuilder();
- for (PipeTaskMeta pipeTaskMeta : runtimeMeta.getConsensusGroupIdToTaskMetaMap().values()) {
- for (Exception e : pipeTaskMeta.getExceptionMessages()) {
- exceptionMessageBuilder.append(e.getMessage()).append("\n");
+ for (PipeRuntimeException e : runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap().values()) {
+ exceptionMessageBuilder
+ .append(DateTimeUtils.convertLongToDate(e.getTimeStamp()))
Review Comment:
```suggestion
.append(DateTimeUtils.convertLongToDate(e.getTimeStamp(), "ms"))
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java:
##########
@@ -105,9 +107,20 @@ public TShowPipeResp convertToTShowPipeResp() {
final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
final StringBuilder exceptionMessageBuilder = new StringBuilder();
- for (PipeTaskMeta pipeTaskMeta : runtimeMeta.getConsensusGroupIdToTaskMetaMap().values()) {
- for (Exception e : pipeTaskMeta.getExceptionMessages()) {
- exceptionMessageBuilder.append(e.getMessage()).append("\n");
+ for (PipeRuntimeException e : runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap().values()) {
+ exceptionMessageBuilder
+ .append(DateTimeUtils.convertLongToDate(e.getTimeStamp()))
+ .append(", ")
+ .append(e.getMessage())
+ .append("\n");
+ }
+ for (PipeTaskMeta pipeTaskMeta : runtimeMeta.getConsensusGroupId2TaskMetaMap().values()) {
+ for (PipeRuntimeException e : pipeTaskMeta.getExceptionMessages()) {
+ exceptionMessageBuilder
+ .append(DateTimeUtils.convertLongToDate(e.getTimeStamp()))
Review Comment:
```suggestion
.append(DateTimeUtils.convertLongToDate(e.getTimeStamp(), "ms"))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252554534
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java:
##########
@@ -93,12 +92,12 @@ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
throws PipeException, IOException {
LOGGER.info("StartPipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName);
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(pipeName, pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to start pipe %s on data nodes. Failures: %s",
- pipeName, result.getMessage()));
+ "Failed to start pipe for pipe %s, details: %s", pipeName, exceptionMessage));
Review Comment:
```suggestion
"Failed to start pipe %s, details: %s", pipeName, exceptionMessage));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253898127
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java:
##########
@@ -930,17 +931,25 @@ public TSStatus alterView(TAlterViewReq req) {
}
@Override
- public TSStatus pushPipeMeta(TPushPipeMetaReq req) {
+ public TPushPipeMetaResp pushPipeMeta(TPushPipeMetaReq req) {
final List<PipeMeta> pipeMetas = new ArrayList<>();
for (ByteBuffer byteBuffer : req.getPipeMetas()) {
pipeMetas.add(PipeMeta.deserialize(byteBuffer));
}
try {
- PipeAgent.task().handlePipeMetaChanges(pipeMetas);
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
+ PipeAgent.task().handlePipeMetaChanges(pipeMetas);
+
+ return exceptionMessages.isEmpty()
+ ? new TPushPipeMetaResp()
+ .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))
+ : new TPushPipeMetaResp()
+ .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+ .setExceptionMessages(exceptionMessages);
} catch (Exception e) {
LOGGER.error("Error occurred when pushing pipe meta", e);
- return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
+ return new TPushPipeMetaResp()
Review Comment:
handle it in parsePushPipeMetaExceptionForPipe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253898590
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/AliasIndexPage.java:
##########
@@ -274,21 +278,20 @@ public String splitByKey(String key, String entry, ByteBuffer dstBuffer, boolean
int sp; // virtual index to split
if (monotonic) {
- // new entry into part with more space
- sp = key.compareTo(lastKey) > 0 ? Math.max(pos, n / 2) : Math.min(pos, n / 2);
+ // New entry into part with more space
+ sp = key.compareTo(lastKey) > 0 ? Math.max(pos, n >> 1) : Math.min(pos, n >> 1);
} else {
- sp = n / 2;
+ sp = n >> 1;
}
- // little different from InternalSegment, only the front edge key can not split
+ // Little different from InternalSegment, only the front edge key can not split
sp = sp <= 0 ? 1 : sp;
- // this method BREAKS envelop of the passing in buffer to be more efficient
+ // This method BREAKS envelop of the passing in buffer to be more efficient
// attributes for dstBuffer
- short memberNum = 0,
- spareOffset = (short) dstBuffer.capacity(),
- spareSize = (short) (spareOffset - SchemaFileConfig.PAGE_HEADER_SIZE);
- long nextSegAddress = this.nextPage;
Review Comment:
long nextSegAddress = this.nextPage; is absent?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253889925
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/handler/AuthorizationHandler.java:
##########
@@ -17,35 +17,24 @@
package org.apache.iotdb.db.protocol.rest.handler;
-import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.rpc.TSStatusCode;
import javax.ws.rs.core.Response;
-import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.SecurityContext;
public class AuthorizationHandler {
public Response checkAuthority(SecurityContext securityContext, Statement statement) {
String userName = securityContext.getUserPrincipal().getName();
- try {
- if (!AuthorityChecker.checkAuthorization(statement, userName)) {
- return Response.ok()
- .entity(
- new ExecutionStatus()
- .code(TSStatusCode.NO_PERMISSION.getStatusCode())
- .message(TSStatusCode.NO_PERMISSION.name()))
- .build();
- }
- } catch (AuthException e) {
+ if (!AuthorityChecker.checkAuthorization(statement, userName)) {
Review Comment:
Do change the code logic here, it was meaningful...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1254006659
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,14 +215,67 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+
+ if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
+ if (!resp.isSetExceptionMessages()) {
+ exceptionMessageBuilder.append(
+ String.format(
+ "DataNodeId: %s, Message: Internal error while processing pushPipeMeta on dataNodes.",
+ dataNodeId));
+ continue;
+ }
+
+ AtomicBoolean hasException = new AtomicBoolean(false);
+
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ // Ignore the timeStamp for simplicity
+ if (pipeName == null) {
+ hasException.set(true);
+ exceptionMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s ",
+ message.getPipeName(), message.getMessage()));
+ } else if (pipeName.equals(message.getPipeName())) {
+ exceptionMessageBuilder.append(
+ String.format("Message: %s ", message.getMessage()));
Review Comment:
```suggestion
} else if (pipeName.equals(message.getPipeName())) {
hasException.set(true);
exceptionMessageBuilder.append(
String.format("Message: %s ", message.getMessage()));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252538183
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,71 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ * @return true if there are exceptions cleared
+ */
+ public boolean clearExceptions(String pipeName) {
+ AtomicBoolean isCleared = new AtomicBoolean(false);
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ runtimeMeta.setExceptionsClearTime(System.currentTimeMillis());
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ exceptionMap.clear();
+ isCleared.set(true);
+ }
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ isCleared.set(true);
+ pipeTaskMeta.clearExceptionMessages();
+ }
+ });
+ return isCleared.get();
+ }
+
+ /**
+ * Record the exceptions of all pipes locally if they encountered failure when pushing pipe meta.
+ * If there are exceptions recorded, the related pipes will be stopped, and the exception messages
+ * will then be updated to all the nodes through PipeHandleMetaChangeProcedure.
+ *
+ * @param respMap The responseMap after pushing pipe meta
+ * @return true if there are exceptions encountered
+ */
+ public boolean recordPushPipeMetaExceptions(Map<Integer, TPushPipeMetaResp> respMap) {
+ boolean hasException = false;
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()) {
+ hasException = true;
+ resp.getExceptionMessages()
Review Comment:
check if ExceptionMessages exists
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253933380
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java:
##########
@@ -128,21 +191,31 @@ public boolean equals(Object o) {
}
PipeRuntimeMeta that = (PipeRuntimeMeta) o;
return Objects.equals(status.get().getType(), that.status.get().getType())
- && consensusGroupIdToTaskMetaMap.equals(that.consensusGroupIdToTaskMetaMap);
+ && consensusGroupId2TaskMetaMap.equals(that.consensusGroupId2TaskMetaMap)
+ && dataNodeId2PipeRuntimeExceptionMap.equals(that.dataNodeId2PipeRuntimeExceptionMap)
+ && exceptionsClearTime.get() == that.exceptionsClearTime.get();
}
@Override
public int hashCode() {
- return Objects.hash(status, consensusGroupIdToTaskMetaMap);
+ return Objects.hash(
+ status,
+ consensusGroupId2TaskMetaMap,
+ dataNodeId2PipeRuntimeExceptionMap,
+ exceptionsClearTime.get());
}
@Override
public String toString() {
return "PipeRuntimeMeta{"
+ "status="
+ status
- + ", consensusGroupIdToTaskMetaMap="
- + consensusGroupIdToTaskMetaMap
- + '}';
+ + ", consensusGroupId2TaskMetaMap="
+ + consensusGroupId2TaskMetaMap
+ + ", dataNodeId2PipeMetaExceptionMap="
+ + dataNodeId2PipeRuntimeExceptionMap
+ + ", exceptionsClearTime="
+ + exceptionsClearTime
+ + ", isStoppedByConfigNode=";
Review Comment:
isStoppedByConfigNode
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252553437
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java:
##########
@@ -145,12 +143,14 @@ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
"CreatePipeProcedureV2: executeFromOperateOnDataNodes({})",
createPipeRequest.getPipeName());
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(
+ createPipeRequest.getPipeName(), pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to create pipe %s on data nodes. Failures: %s",
- pipeStaticMeta.getPipeName(), result.getMessage()));
+ "Failed to create pipe for pipe %s, details: %s",
Review Comment:
```suggestion
"Failed to create pipe %s, details: %s",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252554639
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java:
##########
@@ -132,12 +131,13 @@ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
throws PipeException, IOException {
LOGGER.info("StartPipeProcedureV2: rollbackFromOperateOnDataNodes({})", pipeName);
- TSStatus result = pushPipeMetaToDataNodes(env);
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ parsePushPipeMetaExceptionForPipe(pipeName, pushPipeMetaToDataNodes(env));
+ if (!exceptionMessage.isEmpty()) {
throw new PipeException(
String.format(
- "Failed to rollback start pipe %s on data nodes. Failures: %s",
- pipeName, result.getMessage()));
+ "Failed to rollback start pipe for pipe %s, details: %s",
Review Comment:
```suggestion
"Failed to rollback start pipe %s, details: %s",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253888714
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -192,14 +214,50 @@ protected TSStatus pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IO
pipeMetaBinaryList.add(pipeMeta.serialize());
}
- return RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList));
+ return env.pushPipeMetaToDataNodes(pipeMetaBinaryList);
+ }
+
+ /**
+ * Parsing the given pipe's or all pipes' pushPipeMeta exceptions to string.
+ *
+ * @param pipeName The given pipe's pipe name, null if report all pipes' exceptions.
+ * @param respMap The responseMap after pushing pipe meta
+ * @return Error messages for the given pipe after pushing pipe meta
+ */
+ protected String parsePushPipeMetaExceptionForPipe(
+ String pipeName, Map<Integer, TPushPipeMetaResp> respMap) {
+ final StringBuilder exceptionMessageBuilder = new StringBuilder();
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
+ exceptionMessageBuilder.append(String.format("DataNodeId: %s ", dataNodeId));
Review Comment:
if pipeName is given, and the result (pipeName.equals(message.getPipeName())) is always false, the returned value will not be empty, which may be a unexpeceted behaviour.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253895742
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java:
##########
@@ -151,7 +152,7 @@ private IWritableMemChunkGroup createAlignedMemChunkGroupIfNotExistAndGet(
@Override
public void insert(InsertRowNode insertRowNode) {
- // if this insert plan isn't from storage storageengine (mainly from test), we should set a temp
+ // If this insert plan isn't from storage storageEngine (mainly from test), we should set a temp
Review Comment:
```suggestion
// If this insert plan isn't from storage engine (mainly from test), we should set a temp
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java:
##########
@@ -198,7 +199,7 @@ public void insert(InsertRowNode insertRowNode) {
@Override
public void insertAlignedRow(InsertRowNode insertRowNode) {
- // if this insert node isn't from storage storageengine, we should set a temp device id for it
+ // If this insert node isn't from storage storageEngine, we should set a temp device id for it
Review Comment:
```suggestion
// If this insert node isn't from storage engine, we should set a temp device id for it
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1252546055
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -244,6 +255,71 @@ public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Clear the exceptions of a pipe locally after it starts successfully. If there are exceptions
+ * cleared, the messages will then be updated to all the nodes through
+ * PipeHandleMetaChangeProcedure.
+ *
+ * @param pipeName The name of the pipes to be clear exception
+ * @return true if there are exceptions cleared
+ */
+ public boolean clearExceptions(String pipeName) {
+ AtomicBoolean isCleared = new AtomicBoolean(false);
+ PipeRuntimeMeta runtimeMeta = pipeMetaKeeper.getPipeMeta(pipeName).getRuntimeMeta();
+ runtimeMeta.setExceptionsClearTime(System.currentTimeMillis());
+ Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.isEmpty()) {
+ exceptionMap.clear();
+ isCleared.set(true);
+ }
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .values()
+ .forEach(
+ pipeTaskMeta -> {
+ if (pipeTaskMeta.getExceptionMessages().iterator().hasNext()) {
+ isCleared.set(true);
+ pipeTaskMeta.clearExceptionMessages();
+ }
+ });
+ return isCleared.get();
+ }
+
+ /**
+ * Record the exceptions of all pipes locally if they encountered failure when pushing pipe meta.
+ * If there are exceptions recorded, the related pipes will be stopped, and the exception messages
+ * will then be updated to all the nodes through PipeHandleMetaChangeProcedure.
+ *
+ * @param respMap The responseMap after pushing pipe meta
+ * @return true if there are exceptions encountered
+ */
+ public boolean recordPushPipeMetaExceptions(Map<Integer, TPushPipeMetaResp> respMap) {
+ boolean hasException = false;
+ for (Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
+ int dataNodeId = respEntry.getKey();
+ TPushPipeMetaResp resp = respEntry.getValue();
+ if (resp.getStatus().getCode() == TSStatusCode.PUSH_PIPE_META_ERROR.getStatusCode()) {
+ hasException = true;
+ resp.getExceptionMessages()
+ .forEach(
+ message -> {
+ PipeRuntimeMeta runtimeMeta =
+ pipeMetaKeeper.getPipeMeta(message.getPipeName()).getRuntimeMeta();
+ // Mark the status of the pipe with exception as stopped
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta
+ .getDataNodeId2PipeRuntimeExceptionMap()
+ .put(
+ dataNodeId,
+ new PipeRuntimeCriticalException(
+ message.getMessage(), message.getTimeStamp()));
Review Comment:
if there is already an exception recorded.. should we judge the time and record the latest one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] SteveYurongSu commented on a diff in pull request #10433: [IOTDB-6044] Pipe: Upgraded the exception handling logic
Posted by "SteveYurongSu (via GitHub)" <gi...@apache.org>.
SteveYurongSu commented on code in PR #10433:
URL: https://github.com/apache/iotdb/pull/10433#discussion_r1253866996
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java:
##########
@@ -66,7 +66,14 @@ public TSStatus createPipe(TCreatePipeReq req) {
}
public TSStatus startPipe(String pipeName) {
- return configManager.getProcedureManager().startPipe(pipeName);
+ // Whether there are exceptions to clear
+ boolean hasException = pipeTaskInfo.hasExceptions(pipeName);
Review Comment:
```suggestion
lock();
try {
boolean hasException = pipeTaskInfo.hasExceptions(pipeName);
} finally {
unlock();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org