Refactor shared Pipe, consensus and subscription helpers

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/payload/PipeTransferConfigSnapshotSealReq.java

@@ -26,9 +26,9 @@
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
 
 import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
@@ -78,12 +78,8 @@ public static PipeTransferConfigSnapshotSealReq toTPipeTransferReq(
     return (PipeTransferConfigSnapshotSealReq)
         new PipeTransferConfigSnapshotSealReq()
             .convertToTPipeTransferReq(
-                Objects.nonNull(templateFileName)
-                    ? Arrays.asList(snapshotName, templateFileName)
-                    : Collections.singletonList(snapshotName),
-                Objects.nonNull(templateFileName)
-                    ? Arrays.asList(snapshotLength, templateFileLength)
-                    : Collections.singletonList(snapshotLength),
+                buildFileNameList(snapshotName, templateFileName),
+                buildFileLengthList(snapshotLength, templateFileName, templateFileLength),
                 parameters);
   }
 
@@ -123,15 +119,31 @@ public static byte[] toTPipeTransferBytes(
     parameters.put("authUserName", authUserName);
     return new PipeTransferConfigSnapshotSealReq()
         .convertToTPipeTransferSnapshotSealBytes(
-            Objects.nonNull(templateFileName)
-                ? Arrays.asList(snapshotName, templateFileName)
-                : Collections.singletonList(snapshotName),
-            Objects.nonNull(templateFileName)
-                ? Arrays.asList(snapshotLength, templateFileLength)
-                : Collections.singletonList(snapshotLength),
+            buildFileNameList(snapshotName, templateFileName),
+            buildFileLengthList(snapshotLength, templateFileName, templateFileLength),
             parameters);
   }
 
+  private static List<String> buildFileNameList(
+      final String snapshotName, final String templateFileName) {
+    final List<String> fileNameList = new ArrayList<>();
+    fileNameList.add(snapshotName);
+    if (Objects.nonNull(templateFileName)) {
+      fileNameList.add(templateFileName);
+    }
+    return fileNameList;
+  }
+
+  private static List<Long> buildFileLengthList(
+      final long snapshotLength, final String templateFileName, final long templateFileLength) {
+    final List<Long> fileLengthList = new ArrayList<>();
+    fileLengthList.add(snapshotLength);
+    if (Objects.nonNull(templateFileName)) {
+      fileLengthList.add(templateFileLength);
+    }
+    return fileLengthList;
+  }
+
   /////////////////////////////// Object ///////////////////////////////
 
   @Override

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java

@@ -1373,6 +1373,11 @@ public void cleanupRemoteSnapshot(Peer targetPeer) throws ConsensusGroupModifyPe
   }
 
   public void cleanupSnapshot(String snapshotId) throws ConsensusGroupModifyPeerException {
+    cleanupSnapshotWithoutLock(snapshotId);
+  }
+
+  private void cleanupSnapshotWithoutLock(String snapshotId)
+      throws ConsensusGroupModifyPeerException {
     snapshotReceiveFolderMap.remove(snapshotId);
     List<String> allDirs = new ArrayList<>(Collections.singletonList(storageDir));
     allDirs.addAll(recvFolderManager.getFolders());

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java

@@ -68,11 +68,10 @@ public DispatchLogHandler(
   @Override
   public void onComplete(TSyncLogEntriesRes response) {
     recordTransferAttempt(response);
-    if (response.getStatuses().stream()
-        .anyMatch(status -> RetryUtils.needRetryForWrite(status.getCode()))) {
+    if (response.getStatuses().stream().anyMatch(status -> needRetryForSyncLog(status.getCode()))) {
       List<String> retryStatusMessages =
           response.getStatuses().stream()
-              .filter(status -> RetryUtils.needRetryForWrite(status.getCode()))
+              .filter(status -> needRetryForSyncLog(status.getCode()))
               .map(TSStatus::getMessage)
               .collect(Collectors.toList());
 
@@ -116,6 +115,10 @@ public void onComplete(TSyncLogEntriesRes response) {
     logDispatcherThreadMetrics.recordSyncLogTimePerRequest(System.nanoTime() - createTime);
   }
 
+  private boolean needRetryForSyncLog(int statusCode) {
+    return RetryUtils.needRetryForWrite(statusCode);
+  }
+
   @Override
   public void onError(Exception exception) {
     recordTransferAttempt(false, null, exception);

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java

@@ -90,9 +90,7 @@ public TSyncLogEntriesRes syncLogEntries(TSyncLogEntriesReq req) {
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_SYNC_LOG,
               groupId,
               req.getLogEntries().size());
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TSyncLogEntriesRes(Collections.singletonList(status));
     }
     if (impl.isReadOnly()) {
@@ -154,9 +152,7 @@ public TSyncWriterSafeTimeBarrierRes syncWriterSafeTimeBarrier(
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "TSyncWriterSafeTimeBarrierReq");
-      LOGGER.error(message);
-      final TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      final TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TSyncWriterSafeTimeBarrierRes().setStatus(status);
     }
     impl.observeRemoteWriterSafeTimeBarrier(
@@ -180,9 +176,7 @@ public TInactivatePeerRes inactivatePeer(TInactivatePeerReq req) throws TExcepti
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "inactivatePeer");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TInactivatePeerRes(status);
     }
     impl.setActive(false);
@@ -203,9 +197,7 @@ public TActivatePeerRes activatePeer(TActivatePeerReq req) throws TException {
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "inactivatePeer");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TActivatePeerRes(status);
     }
     KillPoint.setKillPoint(DataNodeKillPoints.DESTINATION_ADD_PEER_DONE);
@@ -225,9 +217,7 @@ public TBuildSyncLogChannelRes buildSyncLogChannel(TBuildSyncLogChannelReq req)
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "buildSyncLogChannel");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TBuildSyncLogChannelRes(status);
     }
     TSStatus responseStatus;
@@ -248,9 +238,7 @@ public TRemoveSyncLogChannelRes removeSyncLogChannel(TRemoveSyncLogChannelReq re
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "buildSyncLogChannel");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TRemoveSyncLogChannelRes(status);
     }
     TSStatus responseStatus;
@@ -310,9 +298,7 @@ public TSendSnapshotFragmentRes sendSnapshotFragment(TSendSnapshotFragmentReq re
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "buildSyncLogChannel");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TSendSnapshotFragmentRes(status);
     }
     TSStatus responseStatus;
@@ -338,9 +324,7 @@ public TTriggerSnapshotLoadRes triggerSnapshotLoad(TTriggerSnapshotLoadReq req)
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "buildSyncLogChannel");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TTriggerSnapshotLoadRes(status);
     }
     if (!impl.loadSnapshot(req.snapshotId)) {
@@ -370,9 +354,7 @@ public TCleanupTransferredSnapshotRes cleanupTransferredSnapshot(
               ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
               groupId,
               "buildSyncLogChannel");
-      LOGGER.error(message);
-      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
-      status.setMessage(message);
+      TSStatus status = getConsensusGroupNotExistStatus(message);
       return new TCleanupTransferredSnapshotRes(status);
     }
     TSStatus responseStatus;
@@ -387,5 +369,10 @@ public TCleanupTransferredSnapshotRes cleanupTransferredSnapshot(
     return new TCleanupTransferredSnapshotRes(responseStatus);
   }
 
+  private static TSStatus getConsensusGroupNotExistStatus(String message) {
+    LOGGER.error(message);
+    return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()).setMessage(message);
+  }
+
   public void handleClientExit() {}
 }

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessorTest.java

@@ -0,0 +1,185 @@
+/*
+ * 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.consensus.iot.service;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.consensus.EmptyStateMachine;
+import org.apache.iotdb.consensus.config.ConsensusConfig;
+import org.apache.iotdb.consensus.i18n.ConsensusMessages;
+import org.apache.iotdb.consensus.iot.IoTConsensus;
+import org.apache.iotdb.consensus.iot.thrift.TActivatePeerReq;
+import org.apache.iotdb.consensus.iot.thrift.TBuildSyncLogChannelReq;
+import org.apache.iotdb.consensus.iot.thrift.TCleanupTransferredSnapshotReq;
+import org.apache.iotdb.consensus.iot.thrift.TInactivatePeerReq;
+import org.apache.iotdb.consensus.iot.thrift.TRemoveSyncLogChannelReq;
+import org.apache.iotdb.consensus.iot.thrift.TSendSnapshotFragmentReq;
+import org.apache.iotdb.consensus.iot.thrift.TSyncLogEntriesReq;
+import org.apache.iotdb.consensus.iot.thrift.TSyncWriterSafeTimeBarrierReq;
+import org.apache.iotdb.consensus.iot.thrift.TTriggerSnapshotLoadReq;
+import org.apache.iotdb.consensus.iot.thrift.TWaitReleaseAllRegionRelatedResourceReq;
+import org.apache.iotdb.consensus.iot.thrift.TWaitSyncLogCompleteReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.Collections;
+
+public class IoTConsensusRPCServiceProcessorTest {
+
+  private static final TConsensusGroupId MISSING_GROUP_ID =
+      new TConsensusGroupId(TConsensusGroupType.DataRegion, 1);
+
+  @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  private IoTConsensus consensus;
+  private IoTConsensusRPCServiceProcessor processor;
+
+  @Before
+  public void setUp() throws IOException {
+    consensus =
+        new IoTConsensus(
+            ConsensusConfig.newBuilder()
+                .setThisNodeId(1)
+                .setThisNode(new TEndPoint("127.0.0.1", 0))
+                .setStorageDir(temporaryFolder.newFolder("storage").getAbsolutePath())
+                .setRecvSnapshotDirs(
+                    Collections.singletonList(
+                        temporaryFolder.newFolder("snapshot").getAbsolutePath()))
+                .setConsensusGroupType(TConsensusGroupType.DataRegion)
+                .build(),
+            groupId -> new EmptyStateMachine());
+    processor = new IoTConsensusRPCServiceProcessor(consensus);
+  }
+
+  @After
+  public void tearDown() {
+    consensus.stop();
+  }
+
+  @Test
+  public void testMissingConsensusGroupPreservesStatusAndMessage() throws Exception {
+    assertMissingGroupStatus(
+        processor
+            .syncLogEntries(
+                new TSyncLogEntriesReq()
+                    .setConsensusGroupId(MISSING_GROUP_ID)
+                    .setLogEntries(Collections.emptyList()))
+            .getStatuses()
+            .get(0),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_SYNC_LOG, "DataRegion[1]", 0));
+    assertMissingGroupStatus(
+        processor
+            .syncWriterSafeTimeBarrier(
+                new TSyncWriterSafeTimeBarrierReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "TSyncWriterSafeTimeBarrierReq"));
+    assertMissingGroupStatus(
+        processor
+            .inactivatePeer(new TInactivatePeerReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "inactivatePeer"));
+    assertMissingGroupStatus(
+        processor
+            .activatePeer(new TActivatePeerReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "inactivatePeer"));
+    assertMissingGroupStatus(
+        processor
+            .buildSyncLogChannel(
+                new TBuildSyncLogChannelReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "buildSyncLogChannel"));
+    assertMissingGroupStatus(
+        processor
+            .removeSyncLogChannel(
+                new TRemoveSyncLogChannelReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "buildSyncLogChannel"));
+    assertMissingGroupStatus(
+        processor
+            .sendSnapshotFragment(
+                new TSendSnapshotFragmentReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "buildSyncLogChannel"));
+    assertMissingGroupStatus(
+        processor
+            .triggerSnapshotLoad(
+                new TTriggerSnapshotLoadReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "buildSyncLogChannel"));
+    assertMissingGroupStatus(
+        processor
+            .cleanupTransferredSnapshot(
+                new TCleanupTransferredSnapshotReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .getStatus(),
+        String.format(
+            ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST,
+            "DataRegion[1]",
+            "buildSyncLogChannel"));
+  }
+
+  @Test
+  public void testMissingConsensusGroupCompletesIdempotentWaits() throws Exception {
+    Assert.assertTrue(
+        processor.waitSyncLogComplete(
+                new TWaitSyncLogCompleteReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .complete);
+    Assert.assertTrue(
+        processor.waitReleaseAllRegionRelatedResource(
+                new TWaitReleaseAllRegionRelatedResourceReq().setConsensusGroupId(MISSING_GROUP_ID))
+            .releaseAllResource);
+  }
+
+  private static void assertMissingGroupStatus(TSStatus status, String expectedMessage) {
+    Assert.assertEquals(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(), status.getCode());
+    Assert.assertEquals(expectedMessage, status.getMessage());
+  }
+}