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 2024/01/18 11:57:50 UTC
(iotdb) branch pipe-retry created (now 2fb53a78ee9)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-retry
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 2fb53a78ee9 Pipe: let non enriched event forever retry when exception occurred to avoid subtask fake running status recorded at data node task agent
This branch includes the following new commits:
new 2fb53a78ee9 Pipe: let non enriched event forever retry when exception occurred to avoid subtask fake running status recorded at data node task agent
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Pipe: let non enriched event forever retry when exception occurred to avoid subtask fake running status recorded at data node task agent
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-retry
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 2fb53a78ee9e0fefc910960549a8165becf41286
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Jan 18 19:57:25 2024 +0800
Pipe: let non enriched event forever retry when exception occurred to avoid subtask fake running status recorded at data node task agent
---
.../db/pipe/task/subtask/PipeDataNodeSubtask.java | 101 +++++++++------
.../subtask/connector/PipeConnectorSubtask.java | 140 ++++++++++-----------
2 files changed, 135 insertions(+), 106 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeDataNodeSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeDataNodeSubtask.java
index 1dbbde4ed0b..be3d6015a5d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeDataNodeSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeDataNodeSubtask.java
@@ -41,15 +41,29 @@ public abstract class PipeDataNodeSubtask extends PipeSubtask {
@Override
public synchronized void onFailure(@NotNull Throwable throwable) {
if (isClosed.get()) {
- LOGGER.info("onFailure in pipe subtask, ignored because pipe is dropped.");
+ LOGGER.info("onFailure in pipe subtask, ignored because pipe is dropped.", throwable);
releaseLastEvent(false);
return;
}
+ if (lastEvent instanceof EnrichedEvent) {
+ onEnrichedEventFailure(throwable);
+ } else {
+ onNonEnrichedEventFailure(throwable);
+ }
+
+ // Although the pipe task will be stopped, we still don't release the last event here
+ // Because we need to keep it for the next retry. If user wants to restart the task,
+ // the last event will be processed again. The last event will be released when the task
+ // is dropped or the process is running normally.
+ }
+
+ private void onEnrichedEventFailure(@NotNull Throwable throwable) {
if (retryCount.get() == 0) {
LOGGER.warn(
- "Failed to execute subtask {}({}), because of {}. Will retry for {} times.",
+ "Failed to execute subtask {} (creation time: {}, simple class: {}), because of {}. Will retry for {} times.",
taskID,
+ creationTime,
this.getClass().getSimpleName(),
throwable.getMessage(),
MAX_RETRY_TIMES,
@@ -59,8 +73,9 @@ public abstract class PipeDataNodeSubtask extends PipeSubtask {
if (retryCount.get() < MAX_RETRY_TIMES) {
retryCount.incrementAndGet();
LOGGER.warn(
- "Retry executing subtask {}({}), retry count [{}/{}]",
+ "Retry executing subtask {} (creation time: {}, simple class: {}), retry count [{}/{}]",
taskID,
+ creationTime,
this.getClass().getSimpleName(),
retryCount.get(),
MAX_RETRY_TIMES);
@@ -68,9 +83,11 @@ public abstract class PipeDataNodeSubtask extends PipeSubtask {
Thread.sleep(1000L * retryCount.get());
} catch (InterruptedException e) {
LOGGER.warn(
- "Interrupted when retrying to execute subtask {}({})",
+ "Interrupted when retrying to execute subtask {} (creation time: {}, simple class: {})",
taskID,
- this.getClass().getSimpleName());
+ creationTime,
+ this.getClass().getSimpleName(),
+ e);
Thread.currentThread().interrupt();
}
@@ -78,47 +95,59 @@ public abstract class PipeDataNodeSubtask extends PipeSubtask {
} else {
final String errorMessage =
String.format(
- "Failed to execute subtask %s(%s), "
+ "Failed to execute subtask %s (creation time: %s, simple class: %s), "
+ "retry count exceeds the max retry times %d, last exception: %s, root cause: %s",
taskID,
+ creationTime,
this.getClass().getSimpleName(),
retryCount.get(),
throwable.getMessage(),
ErrorHandlingUtils.getRootCause(throwable).getMessage());
LOGGER.warn(errorMessage, throwable);
+ ((EnrichedEvent) lastEvent)
+ .reportException(
+ throwable instanceof PipeRuntimeException
+ ? (PipeRuntimeException) throwable
+ : new PipeRuntimeCriticalException(errorMessage));
+ LOGGER.warn(
+ "The last event is an instance of EnrichedEvent, so the exception is reported. "
+ + "Stopping current pipe subtask {} (creation time: {}, simple class: {}) locally... "
+ + "Status shown when query the pipe will be 'STOPPED'. "
+ + "Please restart the task by executing 'START PIPE' manually if needed.",
+ taskID,
+ creationTime,
+ this.getClass().getSimpleName(),
+ throwable);
+ }
+ }
- if (lastEvent instanceof EnrichedEvent) {
- ((EnrichedEvent) lastEvent)
- .reportException(
- throwable instanceof PipeRuntimeException
- ? (PipeRuntimeException) throwable
- : new PipeRuntimeCriticalException(errorMessage));
- LOGGER.warn(
- "The last event is an instance of EnrichedEvent, so the exception is reported. "
- + "Stopping current pipe task {}({}) locally... "
- + "Status shown when query the pipe will be 'STOPPED'. "
- + "Please restart the task by executing 'START PIPE' manually if needed.",
- taskID,
- this.getClass().getSimpleName(),
- throwable);
- } else {
- LOGGER.error(
- "The last event is not an instance of EnrichedEvent, "
- + "so the exception cannot be reported. "
- + "Stopping current pipe task {}({}) locally... "
- + "Status shown when query the pipe will be 'RUNNING' "
- + "instead of 'STOPPED', but the task is actually stopped. "
- + "Please restart the task by executing 'START PIPE' manually if needed.",
- taskID,
- this.getClass().getSimpleName(),
- throwable);
- }
+ private void onNonEnrichedEventFailure(@NotNull Throwable throwable) {
+ if (retryCount.get() == 0) {
+ LOGGER.warn(
+ "Failed to execute subtask {}({}), because of {}. Will retry forever.",
+ taskID,
+ this.getClass().getSimpleName(),
+ throwable.getMessage(),
+ throwable);
+ }
- // Although the pipe task will be stopped, we still don't release the last event here
- // Because we need to keep it for the next retry. If user wants to restart the task,
- // the last event will be processed again. The last event will be released when the task
- // is dropped or the process is running normally.
+ retryCount.incrementAndGet();
+ LOGGER.warn(
+ "Retry executing subtask {}({}), retry count {}",
+ taskID,
+ this.getClass().getSimpleName(),
+ retryCount.get());
+ try {
+ Thread.sleep(Math.min(1000L * retryCount.get(), 10000));
+ } catch (InterruptedException e) {
+ LOGGER.warn(
+ "Interrupted when retrying to execute subtask {}({})",
+ taskID,
+ this.getClass().getSimpleName());
+ Thread.currentThread().interrupt();
}
+
+ submitSelf();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
index 184c706be1c..1ad1f296452 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
@@ -205,86 +205,86 @@ public class PipeConnectorSubtask extends PipeDataNodeSubtask {
return;
}
- // Retry to connect to the target system if the connection is broken
if (throwable instanceof PipeConnectionException) {
- LOGGER.warn(
- "PipeConnectionException occurred, retrying to connect to the target system...",
- throwable);
+ // Retry to connect to the target system if the connection is broken
+ if (onPipeConnectionException(throwable)) {
+ // return if the pipe task should be stopped
+ return;
+ }
+ }
- int retry = 0;
- while (retry < MAX_RETRY_TIMES) {
+ // Handle other exceptions as usual
+ super.onFailure(throwable);
+ }
+
+ /** @return true if the pipe task should be stopped, false otherwise */
+ private boolean onPipeConnectionException(Throwable throwable) {
+ LOGGER.warn(
+ "PipeConnectionException occurred, {} retries to handshake with the target system.",
+ outputPipeConnector.getClass().getName(),
+ throwable);
+
+ int retry = 0;
+ while (retry < MAX_RETRY_TIMES) {
+ try {
+ outputPipeConnector.handshake();
+ LOGGER.info(
+ "{} handshakes with the target system successfully.",
+ outputPipeConnector.getClass().getName());
+ break;
+ } catch (Exception e) {
+ retry++;
+ LOGGER.warn(
+ "{} failed to handshake with the target system for {} times, "
+ + "will retry at most {} times.",
+ outputPipeConnector.getClass().getName(),
+ retry,
+ MAX_RETRY_TIMES,
+ e);
try {
- outputPipeConnector.handshake();
- LOGGER.info("Successfully reconnected to the target system.");
- break;
- } catch (Exception e) {
- retry++;
- LOGGER.warn(
- "Failed to reconnect to the target system, retrying ... "
- + "after [{}/{}] time(s) retries.",
- retry,
- MAX_RETRY_TIMES,
- e);
- try {
- Thread.sleep(retry * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
- } catch (InterruptedException interruptedException) {
- LOGGER.info(
- "Interrupted while sleeping, perhaps need to check "
- + "whether the thread is interrupted.",
- interruptedException);
- Thread.currentThread().interrupt();
- }
+ Thread.sleep(retry * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
+ } catch (InterruptedException interruptedException) {
+ LOGGER.info(
+ "Interrupted while sleeping, will retry to handshake with the target system.",
+ interruptedException);
+ Thread.currentThread().interrupt();
}
}
+ }
- // Stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
- // times
- if (retry == MAX_RETRY_TIMES) {
- if (lastEvent instanceof EnrichedEvent) {
- LOGGER.warn(
- "Failed to reconnect to the target system after {} times, "
- + "stopping current pipe task {}... "
- + "Status shown when query the pipe will be 'STOPPED'. "
- + "Please restart the task by executing 'START PIPE' manually if needed.",
- MAX_RETRY_TIMES,
- taskID,
- throwable);
-
- ((EnrichedEvent) lastEvent)
- .reportException(
- new PipeRuntimeConnectorCriticalException(
- throwable.getMessage()
- + ", root cause: "
- + ErrorHandlingUtils.getRootCause(throwable).getMessage()));
- } else {
- LOGGER.error(
- "Failed to reconnect to the target system after {} times, "
- + "stopping current pipe task {} locally... "
- + "Status shown when query the pipe will be 'RUNNING' instead of 'STOPPED', "
- + "but the task is actually stopped. "
- + "Please restart the task by executing 'START PIPE' manually if needed.",
- MAX_RETRY_TIMES,
- taskID,
- throwable);
- }
-
- // Although the pipe task will be stopped, we still don't release the last event here
- // Because we need to keep it for the next retry. If user wants to restart the task,
- // the last event will be processed again. The last event will be released when the task
- // is dropped or the process is running normally.
-
- // Stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
- return;
- }
- } else {
+ // Stop current pipe task if failed to reconnect to
+ // the target system after MAX_RETRY_TIMES times
+ if (retry == MAX_RETRY_TIMES && lastEvent instanceof EnrichedEvent) {
+ ((EnrichedEvent) lastEvent)
+ .reportException(
+ new PipeRuntimeConnectorCriticalException(
+ throwable.getMessage()
+ + ", root cause: "
+ + ErrorHandlingUtils.getRootCause(throwable).getMessage()));
LOGGER.warn(
- "A non-PipeConnectionException occurred, exception message: {}",
- throwable.getMessage(),
+ "{} failed to handshake with the target system after {} times, "
+ + "stopping current subtask {} (creation time: {}, simple class: {}). "
+ + "Status shown when query the pipe will be 'STOPPED'. "
+ + "Please restart the task by executing 'START PIPE' manually if needed.",
+ outputPipeConnector.getClass().getName(),
+ MAX_RETRY_TIMES,
+ taskID,
+ creationTime,
+ this.getClass().getSimpleName(),
throwable);
+
+ // Although the pipe task will be stopped, we still don't release the last event here
+ // Because we need to keep it for the next retry. If user wants to restart the task,
+ // the last event will be processed again. The last event will be released when the task
+ // is dropped or the process is running normally.
+
+ // Stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
+ return true;
}
- // Handle other exceptions as usual
- super.onFailure(new PipeRuntimeConnectorCriticalException(throwable.getMessage()));
+ // For non enriched event, forever retry.
+ // For enriched event, retry if connection is set up successfully.
+ return false;
}
/**