diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/DataNodeTransportFailureUtils.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/DataNodeTransportFailureUtils.java new file mode 100644 index 000000000000..13b7fda78ca3 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/DataNodeTransportFailureUtils.java @@ -0,0 +1,92 @@ +/* + * 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; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.thrift.TApplicationException; +import org.apache.thrift.protocol.TProtocolException; +import org.apache.thrift.transport.TTransportException; + +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.nio.channels.ClosedChannelException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeoutException; + +public final class DataNodeTransportFailureUtils { + + private static final int TRANSPORT_FAILURE_STATUS_CODE = + TSStatusCode.CAN_NOT_CONNECT_DATANODE.getStatusCode(); + + private DataNodeTransportFailureUtils() {} + + public static boolean isTransportException(final Exception exception) { + final Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable cause = exception; + cause != null && visited.add(cause); + cause = cause.getCause()) { + if (cause instanceof TApplicationException || cause instanceof TProtocolException) { + return false; + } + if (cause instanceof TTransportException) { + final int type = ((TTransportException) cause).getType(); + // Corrupt frames and opening an already-open transport are not connection failures. + return type == TTransportException.UNKNOWN + || type == TTransportException.NOT_OPEN + || type == TTransportException.TIMED_OUT + || type == TTransportException.END_OF_FILE; + } + if (cause instanceof SocketException + || cause instanceof SocketTimeoutException + || cause instanceof ClosedChannelException + || cause instanceof TimeoutException) { + return true; + } + } + return false; + } + + public static TSStatus markTransportFailure(final TSStatus status) { + // CN-local provenance: preserve the metadata error code used by existing callers. + final List subStatus = new ArrayList<>(); + if (status.isSetSubStatus() && status.getSubStatus() != null) { + subStatus.addAll(status.getSubStatus()); + } + subStatus.add(new TSStatus(TRANSPORT_FAILURE_STATUS_CODE)); + status.setSubStatus(subStatus); + return status; + } + + public static boolean isTransportFailure(final TSStatus status) { + return status != null + && status.isSetSubStatus() + && status.getSubStatus() != null + && status.getSubStatus().stream() + .anyMatch( + subStatus -> + subStatus != null && subStatus.getCode() == TRANSPORT_FAILURE_STATUS_CODE); + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/PipePushMetaRPCHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/PipePushMetaRPCHandler.java index 41186a728876..7558d58ef64e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/PipePushMetaRPCHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/PipePushMetaRPCHandler.java @@ -20,7 +20,9 @@ package org.apache.iotdb.confignode.client.async.handlers.rpc; import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; +import org.apache.iotdb.confignode.client.async.DataNodeTransportFailureUtils; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; import org.apache.iotdb.rpc.RpcUtils; @@ -78,9 +80,11 @@ public void onError(final Exception e) { String.valueOf(e.getMessage())); LOGGER.warn(errorMsg); - responseMap.put( - requestId, - new TPushPipeMetaResp(RpcUtils.getStatus(TSStatusCode.PIPE_PUSH_META_ERROR, errorMsg))); + final TSStatus status = RpcUtils.getStatus(TSStatusCode.PIPE_PUSH_META_ERROR, errorMsg); + if (DataNodeTransportFailureUtils.isTransportException(e)) { + DataNodeTransportFailureUtils.markTransportFailure(status); + } + responseMap.put(requestId, new TPushPipeMetaResp(status)); // Always CountDown countDownLatch.countDown(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/subscription/ConsumerGroupPushMetaRPCHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/subscription/ConsumerGroupPushMetaRPCHandler.java index 00a372fd41f9..c6f46468f9d4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/subscription/ConsumerGroupPushMetaRPCHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/subscription/ConsumerGroupPushMetaRPCHandler.java @@ -20,7 +20,9 @@ package org.apache.iotdb.confignode.client.async.handlers.rpc.subscription; import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; +import org.apache.iotdb.confignode.client.async.DataNodeTransportFailureUtils; import org.apache.iotdb.confignode.client.async.handlers.rpc.DataNodeAsyncRequestRPCHandler; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp; @@ -77,10 +79,11 @@ public void onError(Exception e) { String.valueOf(e.getMessage())); LOGGER.warn(errorMsg); - responseMap.put( - requestId, - new TPushConsumerGroupMetaResp( - RpcUtils.getStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR, errorMsg))); + final TSStatus status = RpcUtils.getStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR, errorMsg); + if (DataNodeTransportFailureUtils.isTransportException(e)) { + DataNodeTransportFailureUtils.markTransportFailure(status); + } + responseMap.put(requestId, new TPushConsumerGroupMetaResp(status)); countDownLatch.countDown(); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java index f9e52eb1caad..8fd27ab0fc09 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java @@ -35,6 +35,7 @@ import org.apache.iotdb.confignode.client.CnToCnNodeRequestType; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; +import org.apache.iotdb.confignode.client.async.DataNodeTransportFailureUtils; import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; import org.apache.iotdb.confignode.client.sync.SyncConfigNodeClientPool; import org.apache.iotdb.confignode.consensus.request.write.confignode.RemoveConfigNodePlan; @@ -1099,11 +1100,12 @@ private static void fillMissingPipePushMetaResponses( .putIfAbsent( dataNodeId, new TPushPipeMetaResp( - createMetadataTimeoutStatus( - clientHandler.getRequestType(), - dataNodeId, - TSStatusCode.PIPE_PUSH_META_ERROR, - timeoutInMs)))); + DataNodeTransportFailureUtils.markTransportFailure( + createMetadataTimeoutStatus( + clientHandler.getRequestType(), + dataNodeId, + TSStatusCode.PIPE_PUSH_META_ERROR, + timeoutInMs))))); } private static void fillMissingTopicPushMetaResponses( @@ -1137,11 +1139,12 @@ private static void fillMissingConsumerGroupPushMetaResponses( .putIfAbsent( dataNodeId, new TPushConsumerGroupMetaResp( - createMetadataTimeoutStatus( - clientHandler.getRequestType(), - dataNodeId, - TSStatusCode.CONSUMER_PUSH_META_ERROR, - timeoutInMs)))); + DataNodeTransportFailureUtils.markTransportFailure( + createMetadataTimeoutStatus( + clientHandler.getRequestType(), + dataNodeId, + TSStatusCode.CONSUMER_PUSH_META_ERROR, + timeoutInMs))))); } private static void fillMissingPullCommitProgressResponses( diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java index 579ddd5aafbf..258e68be3416 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java @@ -19,8 +19,10 @@ package org.apache.iotdb.confignode.procedure.impl.subscription; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.confignode.client.async.DataNodeTransportFailureUtils; import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; @@ -333,6 +335,18 @@ protected abstract void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv protected abstract void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws SubscriptionException, IOException; + protected static boolean isOfflineDataNodeTransportFailure( + final ConfigNodeProcedureEnv env, + final int dataNodeId, + final TSStatus status, + final int metadataPushErrorCode, + final boolean hasExceptionMessages) { + return status.getCode() == metadataPushErrorCode + && DataNodeTransportFailureUtils.isTransportFailure(status) + && !hasExceptionMessages + && env.getConfigManager().getLoadManager().getNodeStatus(dataNodeId).isOffline(); + } + /** * Pushing all the topicMeta's to all the dataNodes. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java index 7222b1963be5..330f8010637b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java @@ -127,6 +127,17 @@ public void executeFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) @Override public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws SubscriptionException, IOException { + executeFromOperateOnDataNodes(env, false); + } + + public void executeFromOperateOnDataNodesForUnsubscribe(ConfigNodeProcedureEnv env) + throws SubscriptionException, IOException { + executeFromOperateOnDataNodes(env, true); + } + + private void executeFromOperateOnDataNodes( + ConfigNodeProcedureEnv env, boolean ignoreOfflineDataNodeTransportFailures) + throws SubscriptionException, IOException { LOGGER.info( ProcedureMessages.ALTERCONSUMERGROUPPROCEDURE_EXECUTEFROMOPERATEONDATANODES, updatedConsumerGroupMeta.getConsumerGroupId()); @@ -138,7 +149,19 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) .getSubscriptionReadyDataNodeTracker() .recordSinglePush(existingConsumerGroupMeta, updatedConsumerGroupMeta, responseMap); final List statuses = - responseMap.values().stream().map(TPushConsumerGroupMetaResp::getStatus).toList(); + responseMap.entrySet().stream() + .filter( + entry -> + !ignoreOfflineDataNodeTransportFailures + || !isOfflineDataNodeTransportFailure( + env, + entry.getKey(), + entry.getValue().getStatus(), + TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode(), + entry.getValue().isSetExceptionMessages() + && !entry.getValue().getExceptionMessages().isEmpty())) + .map(entry -> entry.getValue().getStatus()) + .toList(); if (RpcUtils.squashResponseStatusList(statuses).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { // throw exception instead of logging warn, do not rely on metadata synchronization @@ -188,6 +211,17 @@ public void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) @Override public void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws SubscriptionException, IOException { + rollbackFromOperateOnDataNodes(env, false); + } + + public void rollbackFromOperateOnDataNodesForUnsubscribe(ConfigNodeProcedureEnv env) + throws SubscriptionException, IOException { + rollbackFromOperateOnDataNodes(env, true); + } + + private void rollbackFromOperateOnDataNodes( + ConfigNodeProcedureEnv env, boolean ignoreOfflineDataNodeTransportFailures) + throws SubscriptionException, IOException { LOGGER.info(ProcedureMessages.ALTERCONSUMERGROUPPROCEDURE_ROLLBACKFROMOPERATEONDATANODES); final Map responseMap = @@ -197,7 +231,19 @@ public void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) .getSubscriptionReadyDataNodeTracker() .recordSinglePush(updatedConsumerGroupMeta, existingConsumerGroupMeta, responseMap); final List statuses = - responseMap.values().stream().map(TPushConsumerGroupMetaResp::getStatus).toList(); + responseMap.entrySet().stream() + .filter( + entry -> + !ignoreOfflineDataNodeTransportFailures + || !isOfflineDataNodeTransportFailure( + env, + entry.getKey(), + entry.getValue().getStatus(), + TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode(), + entry.getValue().isSetExceptionMessages() + && !entry.getValue().getExceptionMessages().isEmpty())) + .map(entry -> entry.getValue().getStatus()) + .toList(); if (RpcUtils.squashResponseStatusList(statuses).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { // throw exception instead of logging warn, do not rely on metadata synchronization diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java index f2ff6b7301b3..cdb731e1bfa0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java @@ -38,6 +38,7 @@ import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; @@ -49,8 +50,10 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; @@ -190,9 +193,23 @@ protected void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) if (!dropPipeProcedures.isEmpty()) { // An empty Pipe metadata batch triggers full reconciliation and drops every local Pipe. + final Map responseMap = + dropMultiPipeOnDataNodes(dropPipeProcedures, env); + final Map responseMapToCheck = new HashMap<>(responseMap); + responseMapToCheck + .entrySet() + .removeIf( + entry -> + isOfflineDataNodeTransportFailure( + env, + entry.getKey(), + entry.getValue().getStatus(), + TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode(), + entry.getValue().isSetExceptionMessages() + && !entry.getValue().getExceptionMessages().isEmpty())); final String exceptionMessage = AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( - null, dropMultiPipeOnDataNodes(dropPipeProcedures, env)); + null, responseMapToCheck); if (!exceptionMessage.isEmpty()) { // throw exception instead of logging warn, do not rely on metadata synchronization throw new SubscriptionException( @@ -208,7 +225,7 @@ protected void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) } // Push consumer group meta to data nodes - alterConsumerGroupProcedure.executeFromOperateOnDataNodes(env); + alterConsumerGroupProcedure.executeFromOperateOnDataNodesForUnsubscribe(env); } @Override @@ -233,7 +250,7 @@ protected void rollbackFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) LOGGER.info(ProcedureMessages.DROPSUBSCRIPTIONPROCEDURE_ROLLBACKFROMOPERATEONDATANODES); // Rollback AlterConsumerGroupProcedure - alterConsumerGroupProcedure.rollbackFromOperateOnDataNodes(env); + alterConsumerGroupProcedure.rollbackFromOperateOnDataNodesForUnsubscribe(env); // Do nothing to rollback DropPipeProcedureV2s } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java index 7691eb45848e..1d9a6ea76320 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java @@ -19,15 +19,26 @@ package org.apache.iotdb.confignode.procedure.env; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; +import org.apache.iotdb.confignode.client.async.DataNodeTransportFailureUtils; +import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.load.LoadManager; import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler; +import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; +import org.apache.iotdb.rpc.TSStatusCode; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import java.lang.reflect.Method; +import java.util.Map; + import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -90,4 +101,52 @@ public void testNegativeNodeIdIsNotRuntimeActiveWriter() { Assert.assertFalse(env.isRuntimeActiveWriterNode(-1)); } + + @Test + public void testMissingPipeResponsesAreMarkedAsTransportTimeouts() throws Exception { + final DataNodeAsyncRequestContext context = + new DataNodeAsyncRequestContext<>( + CnToDnAsyncRequestType.PIPE_PUSH_MULTI_META, + Map.of(1, new TDataNodeLocation(), 2, new TDataNodeLocation())); + final TPushPipeMetaResp rejected = + new TPushPipeMetaResp(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())); + context.getResponseMap().put(2, rejected); + + fillMissingResponses("fillMissingPipePushMetaResponses", context); + + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportFailure( + context.getResponseMap().get(1).getStatus())); + Assert.assertSame(rejected, context.getResponseMap().get(2)); + Assert.assertFalse(DataNodeTransportFailureUtils.isTransportFailure(rejected.getStatus())); + } + + @Test + public void testMissingConsumerGroupResponsesAreMarkedAsTransportTimeouts() throws Exception { + final DataNodeAsyncRequestContext context = + new DataNodeAsyncRequestContext<>( + CnToDnAsyncRequestType.CONSUMER_GROUP_PUSH_SINGLE_META, + Map.of(1, new TDataNodeLocation(), 2, new TDataNodeLocation())); + final TPushConsumerGroupMetaResp rejected = + new TPushConsumerGroupMetaResp( + new TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode())); + context.getResponseMap().put(2, rejected); + + fillMissingResponses("fillMissingConsumerGroupPushMetaResponses", context); + + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportFailure( + context.getResponseMap().get(1).getStatus())); + Assert.assertSame(rejected, context.getResponseMap().get(2)); + Assert.assertFalse(DataNodeTransportFailureUtils.isTransportFailure(rejected.getStatus())); + } + + private static void fillMissingResponses( + final String methodName, final DataNodeAsyncRequestContext context) throws Exception { + final Method method = + ConfigNodeProcedureEnv.class.getDeclaredMethod( + methodName, DataNodeAsyncRequestContext.class, long.class); + method.setAccessible(true); + method.invoke(null, context, 1L); + } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java index 3dea42d1e82a..aec605b6532c 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java @@ -19,12 +19,21 @@ package org.apache.iotdb.confignode.procedure.impl.subscription.subscription; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; +import org.apache.iotdb.confignode.client.async.DataNodeTransportFailureUtils; +import org.apache.iotdb.confignode.client.async.handlers.rpc.PipePushMetaRPCHandler; +import org.apache.iotdb.confignode.client.async.handlers.rpc.subscription.ConsumerGroupPushMetaRPCHandler; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.subscription.SubscriptionManager; +import org.apache.iotdb.confignode.manager.subscription.SubscriptionReadyDataNodeTracker; import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; @@ -32,10 +41,17 @@ import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.AlterConsumerGroupProcedure; import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaRespExceptionMessage; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; +import org.apache.thrift.TApplicationException; +import org.apache.thrift.protocol.TProtocolException; +import org.apache.thrift.transport.TTransportException; import org.apache.tsfile.utils.PublicBAOS; import org.junit.Assert; import org.junit.Test; @@ -43,6 +59,8 @@ import java.io.DataOutputStream; import java.lang.reflect.Field; +import java.net.ConnectException; +import java.net.SocketTimeoutException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; @@ -51,12 +69,97 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; public class DropSubscriptionProcedureTest { + @Test + public void transportFailureClassificationShouldUseExceptionTypes() { + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportException( + new TTransportException("connection refused"))); + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportException(new ConnectException("connection"))); + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportException(new TimeoutException("timeout"))); + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportException(new SocketTimeoutException("timeout"))); + Assert.assertTrue( + DataNodeTransportFailureUtils.isTransportException( + new RuntimeException(new TimeoutException("timeout")))); + Assert.assertFalse( + DataNodeTransportFailureUtils.isTransportException( + new IllegalStateException("local failure"))); + Assert.assertFalse( + DataNodeTransportFailureUtils.isTransportException( + new TApplicationException("remote application failure"))); + Assert.assertFalse( + DataNodeTransportFailureUtils.isTransportException( + new TProtocolException("invalid reply"))); + Assert.assertFalse( + DataNodeTransportFailureUtils.isTransportException( + new TTransportException(TTransportException.CORRUPTED_DATA, "invalid frame"))); + } + + @Test + public void metadataRpcHandlersShouldMarkOnlyTransportFailures() { + final TDataNodeLocation target = new TDataNodeLocation().setDataNodeId(1); + final Map targetMap = Collections.singletonMap(1, target); + for (final Exception exception : + new Exception[] { + new ConnectException("connection refused"), + new TimeoutException("request timed out"), + new TTransportException(TTransportException.END_OF_FILE, "connection closed"), + new IllegalStateException("local dispatch failure"), + new TApplicationException("remote application failure"), + new TProtocolException("invalid reply"), + new TTransportException(TTransportException.CORRUPTED_DATA, "invalid frame") + }) { + final Map pipeResponses = new HashMap<>(); + new PipePushMetaRPCHandler( + CnToDnAsyncRequestType.PIPE_PUSH_MULTI_META, + 1, + target, + targetMap, + pipeResponses, + new CountDownLatch(1)) + .onError(exception); + final boolean expectedTransportFailure = + exception instanceof ConnectException + || exception instanceof TimeoutException + || (exception instanceof TTransportException + && ((TTransportException) exception).getType() + == TTransportException.END_OF_FILE); + assertEquals( + expectedTransportFailure, + DataNodeTransportFailureUtils.isTransportFailure(pipeResponses.get(1).getStatus())); + assertEquals( + TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode(), + pipeResponses.get(1).getStatus().getCode()); + + final Map consumerGroupResponses = new HashMap<>(); + new ConsumerGroupPushMetaRPCHandler( + CnToDnAsyncRequestType.CONSUMER_GROUP_PUSH_SINGLE_META, + 1, + target, + targetMap, + consumerGroupResponses, + new CountDownLatch(1)) + .onError(exception); + assertEquals( + expectedTransportFailure, + DataNodeTransportFailureUtils.isTransportFailure( + consumerGroupResponses.get(1).getStatus())); + assertEquals( + TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode(), + consumerGroupResponses.get(1).getStatus().getCode()); + } + } + @Test public void serializeDeserializeTest() { PublicBAOS byteArrayOutputStream = new PublicBAOS(); @@ -146,7 +249,263 @@ public void executeFromOperateOnDataNodesShouldSkipEmptyPipeMetaSync() throws Ex Mockito.verify(env, Mockito.never()).pushMultiPipeMetaToDataNodes(Mockito.anyList()); Mockito.verify(env, Mockito.never()).dropMultiPipeOnDataNodes(Mockito.anyList()); - Mockito.verify(alterConsumerGroupProcedure).executeFromOperateOnDataNodes(env); + Mockito.verify(alterConsumerGroupProcedure).executeFromOperateOnDataNodesForUnsubscribe(env); + } + + @Test + public void executeFromOperateOnDataNodesShouldIgnoreOfflinePipeTransportFailure() + throws Exception { + final DropSubscriptionProcedure procedure = new DropSubscriptionProcedure(); + final AlterConsumerGroupProcedure alterConsumerGroupProcedure = + Mockito.mock(AlterConsumerGroupProcedure.class); + procedure.setAlterConsumerGroupProcedure(alterConsumerGroupProcedure); + procedure.setDropPipeProcedures(Collections.singletonList(mockDropPipeProcedure("pipe"))); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Unknown); + Mockito.when(env.dropMultiPipeOnDataNodes(Mockito.anyList())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + transportFailureStatus( + TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode(), "Connection refused")))); + + procedure.executeFromOperateOnDataNodes(env); + + Mockito.verify(alterConsumerGroupProcedure).executeFromOperateOnDataNodesForUnsubscribe(env); + } + + @Test + public void executeFromOperateOnDataNodesShouldNotIgnoreOfflineUnmarkedError() throws Exception { + final DropSubscriptionProcedure procedure = new DropSubscriptionProcedure(); + procedure.setAlterConsumerGroupProcedure(Mockito.mock(AlterConsumerGroupProcedure.class)); + procedure.setDropPipeProcedures(Collections.singletonList(mockDropPipeProcedure("pipe"))); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Unknown); + Mockito.when(env.dropMultiPipeOnDataNodes(Mockito.anyList())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) + .setMessage("local dispatch failure")))); + + try { + procedure.executeFromOperateOnDataNodes(env); + fail("Expected an unmarked request failure to fail the unsubscribe"); + } catch (SubscriptionException e) { + Assert.assertTrue(e.getMessage().contains("local dispatch failure")); + } + } + + @Test + public void executeFromOperateOnDataNodesShouldNotIgnoreOfflinePipeMetaRejection() + throws Exception { + final DropSubscriptionProcedure procedure = new DropSubscriptionProcedure(); + final AlterConsumerGroupProcedure alterConsumerGroupProcedure = + Mockito.mock(AlterConsumerGroupProcedure.class); + procedure.setAlterConsumerGroupProcedure(alterConsumerGroupProcedure); + procedure.setDropPipeProcedures(Collections.singletonList(mockDropPipeProcedure("pipe"))); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Unknown); + Mockito.when(env.dropMultiPipeOnDataNodes(Mockito.anyList())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages( + Collections.singletonList( + new TPushPipeMetaRespExceptionMessage( + "pipe", "metadata rejected", System.currentTimeMillis()))))); + + try { + procedure.executeFromOperateOnDataNodes(env); + fail("Expected the DataNode's metadata rejection to fail the unsubscribe"); + } catch (SubscriptionException e) { + Assert.assertTrue(e.getMessage().contains("metadata rejected")); + } + + Mockito.verify(alterConsumerGroupProcedure, Mockito.never()) + .executeFromOperateOnDataNodesForUnsubscribe(Mockito.any()); + } + + @Test + public void executeFromOperateOnDataNodesShouldNotIgnoreOnlinePipeTransportFailure() + throws Exception { + final DropSubscriptionProcedure procedure = new DropSubscriptionProcedure(); + final AlterConsumerGroupProcedure alterConsumerGroupProcedure = + Mockito.mock(AlterConsumerGroupProcedure.class); + procedure.setAlterConsumerGroupProcedure(alterConsumerGroupProcedure); + procedure.setDropPipeProcedures(Collections.singletonList(mockDropPipeProcedure("pipe"))); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Running); + Mockito.when(env.dropMultiPipeOnDataNodes(Mockito.anyList())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + transportFailureStatus( + TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode(), "Connection refused")))); + + try { + procedure.executeFromOperateOnDataNodes(env); + fail("Expected a transport failure from an online DataNode to fail the unsubscribe"); + } catch (SubscriptionException e) { + Assert.assertTrue(e.getMessage().contains("Connection refused")); + } + + Mockito.verify(alterConsumerGroupProcedure, Mockito.never()) + .executeFromOperateOnDataNodesForUnsubscribe(Mockito.any()); + } + + @Test + public void alterConsumerGroupShouldIgnoreOfflineTransportFailureWhenUnsubscribing() + throws Exception { + final ConsumerGroupMeta existingMeta = + new ConsumerGroupMeta("group", 1, new ConsumerMeta("consumer", 1, Collections.emptyMap())); + existingMeta.addSubscription("consumer", Collections.singleton("topic")); + final ConsumerGroupMeta updatedMeta = existingMeta.deepCopy(); + updatedMeta.removeSubscription("consumer", Collections.singleton("topic")); + final AlterConsumerGroupProcedure procedure = new AlterConsumerGroupProcedure(); + procedure.setExistingConsumerGroupMeta(existingMeta); + procedure.setUpdatedConsumerGroupMeta(updatedMeta); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Unknown); + final ConfigManager configManager = env.getConfigManager(); + final SubscriptionManager subscriptionManager = Mockito.mock(SubscriptionManager.class); + final SubscriptionReadyDataNodeTracker readyDataNodeTracker = + new SubscriptionReadyDataNodeTracker(); + Mockito.when(configManager.getSubscriptionManager()).thenReturn(subscriptionManager); + Mockito.when(subscriptionManager.getSubscriptionReadyDataNodeTracker()) + .thenReturn(readyDataNodeTracker); + readyDataNodeTracker.replaceFromFullPush( + Collections.singletonList(existingMeta), + Collections.singletonMap( + 1, + new TPushConsumerGroupMetaResp( + new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())))); + Mockito.when(env.pushSingleConsumerGroupOnDataNode(Mockito.any(ByteBuffer.class))) + .thenReturn( + Collections.singletonMap( + 1, + new TPushConsumerGroupMetaResp( + transportFailureStatus( + TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode(), + "Connection refused")))); + + procedure.executeFromOperateOnDataNodesForUnsubscribe(env); + procedure.rollbackFromOperateOnDataNodesForUnsubscribe(env); + + assertEquals( + Collections.emptyList(), + readyDataNodeTracker.getReadyDataNodeIds( + "topic", "group", existingMeta.visibleUnder(true), Collections.singleton(1))); + } + + @Test + public void alterConsumerGroupShouldNotIgnoreOfflineUnmarkedErrorWhenUnsubscribing() + throws Exception { + final ConsumerGroupMeta existingMeta = + new ConsumerGroupMeta("group", 1, new ConsumerMeta("consumer", 1, Collections.emptyMap())); + final AlterConsumerGroupProcedure procedure = new AlterConsumerGroupProcedure(); + procedure.setExistingConsumerGroupMeta(existingMeta); + procedure.setUpdatedConsumerGroupMeta(existingMeta.deepCopy()); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Unknown); + final SubscriptionManager subscriptionManager = Mockito.mock(SubscriptionManager.class); + Mockito.when(env.getConfigManager().getSubscriptionManager()).thenReturn(subscriptionManager); + Mockito.when(subscriptionManager.getSubscriptionReadyDataNodeTracker()) + .thenReturn(Mockito.mock(SubscriptionReadyDataNodeTracker.class)); + Mockito.when(env.pushSingleConsumerGroupOnDataNode(Mockito.any(ByteBuffer.class))) + .thenReturn( + Collections.singletonMap( + 1, + new TPushConsumerGroupMetaResp( + new TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode()) + .setMessage("local dispatch failure")))); + + try { + procedure.executeFromOperateOnDataNodesForUnsubscribe(env); + fail("Expected an unmarked request failure to fail the consumer group update"); + } catch (SubscriptionException e) { + Assert.assertTrue(e.getMessage().contains("local dispatch failure")); + } + Assert.assertThrows( + SubscriptionException.class, + () -> procedure.rollbackFromOperateOnDataNodesForUnsubscribe(env)); + } + + @Test + public void alterConsumerGroupShouldNotIgnoreOnlineTransportFailureWhenUnsubscribing() + throws Exception { + final ConsumerGroupMeta existingMeta = + new ConsumerGroupMeta("group", 1, new ConsumerMeta("consumer", 1, Collections.emptyMap())); + final AlterConsumerGroupProcedure procedure = new AlterConsumerGroupProcedure(); + procedure.setExistingConsumerGroupMeta(existingMeta); + procedure.setUpdatedConsumerGroupMeta(existingMeta.deepCopy()); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Running); + final SubscriptionManager subscriptionManager = Mockito.mock(SubscriptionManager.class); + Mockito.when(env.getConfigManager().getSubscriptionManager()).thenReturn(subscriptionManager); + Mockito.when(subscriptionManager.getSubscriptionReadyDataNodeTracker()) + .thenReturn(Mockito.mock(SubscriptionReadyDataNodeTracker.class)); + Mockito.when(env.pushSingleConsumerGroupOnDataNode(Mockito.any(ByteBuffer.class))) + .thenReturn( + Collections.singletonMap( + 1, + new TPushConsumerGroupMetaResp( + transportFailureStatus( + TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode(), + "Connection refused")))); + + try { + procedure.executeFromOperateOnDataNodesForUnsubscribe(env); + fail("Expected a transport failure from an online DataNode to fail the update"); + } catch (SubscriptionException e) { + Assert.assertTrue(e.getMessage().contains("Connection refused")); + } + Assert.assertThrows( + SubscriptionException.class, + () -> procedure.rollbackFromOperateOnDataNodesForUnsubscribe(env)); + } + + @Test + public void alterConsumerGroupShouldNotIgnoreOfflineMetadataRejectionWhenUnsubscribing() + throws Exception { + final ConsumerGroupMeta existingMeta = + new ConsumerGroupMeta("group", 1, new ConsumerMeta("consumer", 1, Collections.emptyMap())); + final ConsumerGroupMeta updatedMeta = existingMeta.deepCopy(); + final AlterConsumerGroupProcedure procedure = new AlterConsumerGroupProcedure(); + procedure.setExistingConsumerGroupMeta(existingMeta); + procedure.setUpdatedConsumerGroupMeta(updatedMeta); + + final ConfigNodeProcedureEnv env = mockEnvWithNodeStatus(NodeStatus.Unknown); + final ConfigManager configManager = env.getConfigManager(); + final SubscriptionManager subscriptionManager = Mockito.mock(SubscriptionManager.class); + Mockito.when(configManager.getSubscriptionManager()).thenReturn(subscriptionManager); + Mockito.when(subscriptionManager.getSubscriptionReadyDataNodeTracker()) + .thenReturn(Mockito.mock(SubscriptionReadyDataNodeTracker.class)); + Mockito.when(env.pushSingleConsumerGroupOnDataNode(Mockito.any(ByteBuffer.class))) + .thenReturn( + Collections.singletonMap( + 1, + new TPushConsumerGroupMetaResp( + new TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages( + Collections.singletonList( + new TPushConsumerGroupMetaRespExceptionMessage( + "group", "metadata rejected", System.currentTimeMillis()))))); + + try { + procedure.executeFromOperateOnDataNodesForUnsubscribe(env); + fail("Expected the DataNode's metadata rejection to fail the consumer group update"); + } catch (SubscriptionException e) { + Assert.assertNotNull(e.getMessage()); + } + Assert.assertThrows( + SubscriptionException.class, + () -> procedure.rollbackFromOperateOnDataNodesForUnsubscribe(env)); } @Test @@ -234,6 +593,27 @@ private static ConfigNodeProcedureEnv mockConsensusFailureEnv(final TSStatus res return env; } + private static ConfigNodeProcedureEnv mockEnvWithNodeStatus(final NodeStatus nodeStatus) { + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final LoadManager loadManager = Mockito.mock(LoadManager.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getLoadManager()).thenReturn(loadManager); + Mockito.when(loadManager.getNodeStatus(1)).thenReturn(nodeStatus); + return env; + } + + private static DropPipeProcedureV2 mockDropPipeProcedure(final String pipeName) { + final DropPipeProcedureV2 procedure = Mockito.mock(DropPipeProcedureV2.class); + Mockito.when(procedure.getPipeName()).thenReturn(pipeName); + return procedure; + } + + private static TSStatus transportFailureStatus(final int statusCode, final String message) { + return DataNodeTransportFailureUtils.markTransportFailure( + new TSStatus(statusCode).setMessage(message)); + } + private static void setField(final Object target, final String fieldName, final Object value) throws Exception { Class clazz = target.getClass();