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/10/12 11:51:32 UTC
(iotdb) branch rc/1.3.3 updated: Subscription: downgrade pipe timeout exception occurred in procedure for subscription-side & enlarge message invisible time default parameter (#13685) (#13749)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/1.3.3 by this push:
new d50dc43137d Subscription: downgrade pipe timeout exception occurred in procedure for subscription-side & enlarge message invisible time default parameter (#13685) (#13749)
d50dc43137d is described below
commit d50dc43137d99561f17bb6ea3c9511e8dfb9236c
Author: V_Galaxy <vg...@apache.org>
AuthorDate: Sat Oct 12 19:51:26 2024 +0800
Subscription: downgrade pipe timeout exception occurred in procedure for subscription-side & enlarge message invisible time default parameter (#13685) (#13749)
---
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
....java => SubscriptionPipeTimeoutException.java} | 13 ++++-----
.../SubscriptionRuntimeCriticalException.java | 2 +-
.../consumer/SubscriptionConsumer.java | 11 +++++++
.../consumer/SubscriptionProvider.java | 34 +++++++++++++++-------
.../consumer/SubscriptionProviders.java | 3 +-
.../iotdb/confignode/manager/ProcedureManager.java | 16 +++++++---
.../receiver/SubscriptionReceiverV1.java | 26 +++++++++++++++--
.../apache/iotdb/commons/conf/CommonConfig.java | 2 +-
9 files changed, 81 insertions(+), 27 deletions(-)
diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 3b7cd2b62ce..e69b980d388 100644
--- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -254,6 +254,7 @@ public enum TSStatusCode {
SUBSCRIPTION_UNSUBSCRIBE_ERROR(1908),
SUBSCRIPTION_MISSING_CUSTOMER(1909),
SHOW_SUBSCRIPTION_ERROR(1910),
+ SUBSCRIPTION_PIPE_TIMEOUT_ERROR(1911),
// Topic
CREATE_TOPIC_ERROR(2000),
diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionPipeTimeoutException.java
similarity index 67%
copy from iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java
copy to iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionPipeTimeoutException.java
index 72124ba3109..f9ecc0f1b8b 100644
--- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionPipeTimeoutException.java
@@ -21,22 +21,21 @@ package org.apache.iotdb.rpc.subscription.exception;
import java.util.Objects;
-public class SubscriptionRuntimeCriticalException extends SubscriptionException {
+public class SubscriptionPipeTimeoutException extends SubscriptionRuntimeNonCriticalException {
- public SubscriptionRuntimeCriticalException(final String message) {
+ public SubscriptionPipeTimeoutException(final String message) {
super(message);
}
- public SubscriptionRuntimeCriticalException(final String message, final Throwable cause) {
+ public SubscriptionPipeTimeoutException(final String message, final Throwable cause) {
super(message, cause);
}
@Override
public boolean equals(final Object obj) {
- return obj instanceof SubscriptionRuntimeCriticalException
- && Objects.equals(getMessage(), ((SubscriptionRuntimeCriticalException) obj).getMessage())
- && Objects.equals(
- getTimeStamp(), ((SubscriptionRuntimeCriticalException) obj).getTimeStamp());
+ return obj instanceof SubscriptionPipeTimeoutException
+ && Objects.equals(getMessage(), ((SubscriptionPipeTimeoutException) obj).getMessage())
+ && Objects.equals(getTimeStamp(), ((SubscriptionPipeTimeoutException) obj).getTimeStamp());
}
@Override
diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java
index 72124ba3109..0e471dca777 100644
--- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionRuntimeCriticalException.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.rpc.subscription.exception;
import java.util.Objects;
-public class SubscriptionRuntimeCriticalException extends SubscriptionException {
+public class SubscriptionRuntimeCriticalException extends SubscriptionRuntimeException {
public SubscriptionRuntimeCriticalException(final String message) {
super(message);
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
index 8cf4213911d..cd7f2b20571 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionConnectionException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionPipeTimeoutException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeNonCriticalException;
import org.apache.iotdb.rpc.subscription.payload.poll.ErrorPayload;
@@ -1211,6 +1212,11 @@ abstract class SubscriptionConsumer implements AutoCloseable {
subscribedTopics = provider.subscribe(topicNames);
return;
} catch (final Exception e) {
+ if (e instanceof SubscriptionPipeTimeoutException) {
+ // degrade exception to log for pipe timeout
+ LOGGER.warn(e.getMessage());
+ return;
+ }
LOGGER.warn(
"{} failed to subscribe topics {} from subscription provider {}, try next subscription provider...",
this,
@@ -1241,6 +1247,11 @@ abstract class SubscriptionConsumer implements AutoCloseable {
subscribedTopics = provider.unsubscribe(topicNames);
return;
} catch (final Exception e) {
+ if (e instanceof SubscriptionPipeTimeoutException) {
+ // degrade exception to log for pipe timeout
+ LOGGER.warn(e.getMessage());
+ return;
+ }
LOGGER.warn(
"{} failed to unsubscribe topics {} from subscription provider {}, try next subscription provider...",
this,
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java
index ab22ba904ef..ef03b6f20cb 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionConnectionException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionPipeTimeoutException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeNonCriticalException;
import org.apache.iotdb.rpc.subscription.payload.poll.PollFilePayload;
@@ -67,6 +68,14 @@ final class SubscriptionProvider extends SubscriptionSession {
private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionProvider.class);
+ private static final String STATUS_FORMATTER = "Status code is [%s], status message is [%s].";
+ private static final String INTERNAL_ERROR_FORMATTER =
+ "Internal error occurred. " + STATUS_FORMATTER;
+ private static final String SUBSCRIPTION_PIPE_TIMEOUT_FORMATTER =
+ "A timeout has occurred in procedures related to the pipe within the subscription procedure. "
+ + "Please manually check the subscription correctness later. "
+ + STATUS_FORMATTER;
+
private String consumerId;
private String consumerGroupId;
@@ -393,20 +402,25 @@ final class SubscriptionProvider extends SubscriptionSession {
case 1906: // SUBSCRIPTION_CLOSE_ERROR
case 1907: // SUBSCRIPTION_SUBSCRIBE_ERROR
case 1908: // SUBSCRIPTION_UNSUBSCRIBE_ERROR
- LOGGER.warn(
- "Internal error occurred, status code {}, status message {}",
- status.code,
- status.message);
- throw new SubscriptionRuntimeNonCriticalException(status.message);
+ {
+ final String errorMessage =
+ String.format(INTERNAL_ERROR_FORMATTER, status.code, status.message);
+ LOGGER.warn(errorMessage);
+ throw new SubscriptionRuntimeNonCriticalException(errorMessage);
+ }
+ case 1911: // SUBSCRIPTION_PIPE_TIMEOUT_ERROR
+ throw new SubscriptionPipeTimeoutException(
+ String.format(SUBSCRIPTION_PIPE_TIMEOUT_FORMATTER, status.code, status.message));
case 1900: // SUBSCRIPTION_VERSION_ERROR
case 1901: // SUBSCRIPTION_TYPE_ERROR
case 1909: // SUBSCRIPTION_MISSING_CUSTOMER
default:
- LOGGER.warn(
- "Internal error occurred, status code {}, status message {}",
- status.code,
- status.message);
- throw new SubscriptionRuntimeCriticalException(status.message);
+ {
+ final String errorMessage =
+ String.format(INTERNAL_ERROR_FORMATTER, status.code, status.message);
+ LOGGER.warn(errorMessage);
+ throw new SubscriptionRuntimeCriticalException(status.message);
+ }
}
}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProviders.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProviders.java
index fa865af3644..4fb93430bc8 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProviders.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProviders.java
@@ -133,7 +133,8 @@ final class SubscriptionProviders {
for (final SubscriptionProvider provider : getAllProviders()) {
try {
provider.close();
- } catch (final Exception ignored) {
+ } catch (final Exception e) {
+ LOGGER.warn(e.getMessage());
}
}
subscriptionProviders.clear();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index 0492ca0aaff..85e4d85e2d7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -1140,9 +1140,13 @@ public class ProcedureManager {
waitingProcedureFinished(Collections.singletonList(procedureId), statusList);
if (isSucceed) {
return statusList.get(0);
- } else {
- return new TSStatus(TSStatusCode.SUBSCRIPTION_SUBSCRIBE_ERROR.getStatusCode())
+ } else if (PROCEDURE_TIMEOUT_MESSAGE.equals(statusList.get(0).getMessage())) {
+ // we assume that a timeout has occurred in the procedure related to the pipe in the
+ // subscription procedure
+ return new TSStatus(TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode())
.setMessage(wrapTimeoutMessageForPipeProcedure(statusList.get(0).getMessage()));
+ } else {
+ return new TSStatus(TSStatusCode.SUBSCRIPTION_SUBSCRIBE_ERROR.getStatusCode());
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.SUBSCRIPTION_SUBSCRIBE_ERROR.getStatusCode())
@@ -1158,9 +1162,13 @@ public class ProcedureManager {
waitingProcedureFinished(Collections.singletonList(procedureId), statusList);
if (isSucceed) {
return statusList.get(0);
- } else {
- return new TSStatus(TSStatusCode.SUBSCRIPTION_UNSUBSCRIBE_ERROR.getStatusCode())
+ } else if (PROCEDURE_TIMEOUT_MESSAGE.equals(statusList.get(0).getMessage())) {
+ // we assume that a timeout has occurred in the procedure related to the pipe in the
+ // subscription procedure
+ return new TSStatus(TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode())
.setMessage(wrapTimeoutMessageForPipeProcedure(statusList.get(0).getMessage()));
+ } else {
+ return new TSStatus(TSStatusCode.SUBSCRIPTION_UNSUBSCRIBE_ERROR.getStatusCode());
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.SUBSCRIPTION_UNSUBSCRIBE_ERROR.getStatusCode())
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
index 907bae11302..97401c73808 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
@@ -40,6 +40,7 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionPayloadExceedException;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionPipeTimeoutException;
import org.apache.iotdb.rpc.subscription.payload.poll.PollFilePayload;
import org.apache.iotdb.rpc.subscription.payload.poll.PollPayload;
import org.apache.iotdb.rpc.subscription.payload.poll.PollTabletsPayload;
@@ -259,6 +260,9 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
private TPipeSubscribeResp handlePipeSubscribeSubscribe(final PipeSubscribeSubscribeReq req) {
try {
return handlePipeSubscribeSubscribeInternal(req);
+ } catch (final SubscriptionPipeTimeoutException e) {
+ return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(
+ RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR, e.getMessage()));
} catch (final Exception e) {
LOGGER.warn("Exception occurred when subscribing with request {}", req, e);
final String exceptionMessage =
@@ -297,6 +301,9 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
private TPipeSubscribeResp handlePipeSubscribeUnsubscribe(final PipeSubscribeUnsubscribeReq req) {
try {
return handlePipeSubscribeUnsubscribeInternal(req);
+ } catch (final SubscriptionPipeTimeoutException e) {
+ return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(
+ RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR, e.getMessage()));
} catch (final Exception e) {
LOGGER.warn("Exception occurred when unsubscribing with request {}", req, e);
final String exceptionMessage =
@@ -571,7 +578,12 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
"Subscription: unsubscribe all subscribed topics {} before close consumer {}",
topicNames,
consumerConfig);
- unsubscribe(consumerConfig, topicNames);
+ try {
+ unsubscribe(consumerConfig, topicNames);
+ } catch (final SubscriptionPipeTimeoutException e) {
+ LOGGER.warn(e.getMessage());
+ // continue drop consumer operation
+ } // rethrow other exceptions
}
// drop consumer if existed
@@ -670,7 +682,11 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
String.format(
"Subscription: Failed to subscribe topics %s for consumer %s in config node, status is %s.",
topicNames, consumerConfig, tsStatus);
- throw new SubscriptionException(exceptionMessage);
+ if (TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode() == tsStatus.getCode()) {
+ throw new SubscriptionPipeTimeoutException(exceptionMessage);
+ } else {
+ throw new SubscriptionException(exceptionMessage);
+ }
}
} catch (final ClientManagerException | TException e) {
LOGGER.warn(
@@ -706,7 +722,11 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
String.format(
"Subscription: Failed to unsubscribe topics %s for consumer %s in config node, status is %s.",
topicNames, consumerConfig, tsStatus);
- throw new SubscriptionException(exceptionMessage);
+ if (TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode() == tsStatus.getCode()) {
+ throw new SubscriptionPipeTimeoutException(exceptionMessage);
+ } else {
+ throw new SubscriptionException(exceptionMessage);
+ }
}
} catch (final ClientManagerException | TException e) {
LOGGER.warn(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 1a5505737cc..46ad1c14bbf 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -283,7 +283,7 @@ public class CommonConfig {
private int subscriptionPollMaxBlockingTimeMs = 500;
private int subscriptionSerializeMaxBlockingTimeMs = 100;
private long subscriptionLaunchRetryIntervalMs = 1000;
- private int subscriptionRecycleUncommittedEventIntervalMs = 60000; // 60s
+ private int subscriptionRecycleUncommittedEventIntervalMs = 600000; // 600s
private long subscriptionReadFileBufferSize = 8 * MB;
private long subscriptionReadTabletBufferSize = 8 * MB;
private long subscriptionTsFileDeduplicationWindowSeconds = 120; // 120s