Fix unsubscribe failures when DataNodes are offline

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<Throwable> 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<TSStatus> 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);
+  }
+}

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();

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();
   }

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(

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.
    *

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java

@@ -127,6 +127,17 @@
   @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 @@
         .getSubscriptionReadyDataNodeTracker()
         .recordSinglePush(existingConsumerGroupMeta, updatedConsumerGroupMeta, responseMap);
     final List<TSStatus> 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 @@
   @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<Integer, TPushConsumerGroupMetaResp> responseMap =
@@ -197,7 +231,19 @@
         .getSubscriptionReadyDataNodeTracker()
         .recordSinglePush(updatedConsumerGroupMeta, existingConsumerGroupMeta, responseMap);
     final List<TSStatus> 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

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<Integer, TPushPipeMetaResp> responseMap =
+          dropMultiPipeOnDataNodes(dropPipeProcedures, env);
+      final Map<Integer, TPushPipeMetaResp> 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
   }