Fix concurrent subscription consumer handshakes

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -332,6 +332,7 @@ public enum TSStatusCode {
   SUBSCRIPTION_OWNER_EPOCH_REQUIRED(1916),
   SUBSCRIPTION_OWNER_LEASE_EXPIRED(1917),
   SUBSCRIPTION_OWNER_EPOCH_CONFLICT(1918),
+  SUBSCRIPTION_CONSUMER_FENCED(1919),
 
   // Topic
   CREATE_TOPIC_ERROR(2000),

iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/ConsumerConfig.java

@@ -68,6 +68,10 @@ public String getConsumerGroupId() {
     return getString(ConsumerConstant.CONSUMER_GROUP_ID_KEY);
   }
 
+  public String getConsumerInstanceId() {
+    return getString(ConsumerConstant.CONSUMER_INSTANCE_ID_KEY);
+  }
+
   public String getOwnerId() {
     return getString(ConsumerConstant.OWNER_ID_KEY);
   }

iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/ConsumerConstant.java

@@ -40,6 +40,7 @@ public class ConsumerConstant {
 
   public static final String CONSUMER_ID_KEY = "consumer-id";
   public static final String CONSUMER_GROUP_ID_KEY = "group-id";
+  public static final String CONSUMER_INSTANCE_ID_KEY = "consumer-instance-id";
   public static final String OWNER_ID_KEY = "owner-id";
   public static final String OWNER_EPOCH_KEY = "owner-epoch";
 

iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionConsumerFencedException.java

@@ -0,0 +1,51 @@
+/*
+ * 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.rpc.subscription.exception;
+
+import java.util.Objects;
+
+/**
+ * Indicates that another connection has taken over the same subscription consumer identity. The
+ * fenced consumer instance can no longer issue requests and must not reclaim ownership
+ * automatically.
+ */
+public class SubscriptionConsumerFencedException extends SubscriptionRuntimeCriticalException {
+
+  public SubscriptionConsumerFencedException(final String message) {
+    super(message);
+  }
+
+  public SubscriptionConsumerFencedException(final String message, final Throwable cause) {
+    super(message, cause);
+  }
+
+  @Override
+  public boolean equals(final Object obj) {
+    return obj instanceof SubscriptionConsumerFencedException
+        && Objects.equals(getMessage(), ((SubscriptionConsumerFencedException) obj).getMessage())
+        && Objects.equals(
+            getTimeStamp(), ((SubscriptionConsumerFencedException) obj).getTimeStamp());
+  }
+
+  @Override
+  public int hashCode() {
+    return super.hashCode();
+  }
+}