Pipe IT: Stabilize table model leader stop test

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java

@@ -313,6 +313,9 @@ public void testPipeAfterDataRegionLeaderStop() {
           TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
 
           TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+          // Avoid electing a stale follower after stopping the current test1 leader.
+          flushTableDataRegionReplicasAfterReplicationComplete(
+              senderEnv, Collections.singletonList("test1"));
 
           final int leaderIndex = restartTableDataRegionLeader(client, "test1");
           if (leaderIndex == -1) { // ensure the leader is stopped
@@ -324,7 +327,10 @@ public void testPipeAfterDataRegionLeaderStop() {
           TableModelUtils.insertData("test1", "test1", 200, 300, senderEnv);
 
           TableModelUtils.assertData("test", "test", 0, 300, receiverEnv, handleFailure);
-          waitForTableDataRegionReplicationComplete(Arrays.asList("test", "test1"));
+          flushTableDataRegionReplicasAfterReplicationComplete(
+              senderEnv, Arrays.asList("test", "test1"));
+          flushTableDataRegionReplicasAfterReplicationComplete(
+              receiverEnv, Collections.singletonList("test"));
         }
 
         try {
@@ -428,14 +434,22 @@ private int restartTableDataRegionLeader(
     return -1;
   }
 
-  private void waitForTableDataRegionReplicationComplete(final List<String> databases) {
+  private void flushTableDataRegionReplicasAfterReplicationComplete(
+      final BaseEnv env, final List<String> databases) {
+    waitForTableDataRegionReplicationComplete(env, databases);
+    TestUtils.executeNonQueryWithRetry(env, "flush");
+    waitForTableDataRegionReplicationComplete(env, databases);
+  }
+
+  private void waitForTableDataRegionReplicationComplete(
+      final BaseEnv env, final List<String> databases) {
     await()
         .pollInterval(500, TimeUnit.MILLISECONDS)
         .atMost(2, TimeUnit.MINUTES)
         .untilAsserted(
             () -> {
               try (final SyncConfigNodeIServiceClient client =
-                  (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+                  (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) {
                 final List<TRegionInfo> leaderRegionInfoList =
                     showTableDataRegionLeaders(databases, client);
                 Assert.assertFalse(
@@ -444,14 +458,14 @@ private void waitForTableDataRegionReplicationComplete(final List<String> databa
 
                 for (final TRegionInfo regionInfo : leaderRegionInfoList) {
                   final DataNodeWrapper leaderNode =
-                      findDataNodeWrapperByPort(regionInfo.getClientRpcPort());
+                      findDataNodeWrapperByPort(env, regionInfo.getClientRpcPort());
                   final String metricsUrl =
                       "http://"
                           + leaderNode.getIp()
                           + ":"
                           + leaderNode.getMetricPort()
                           + "/metrics";
-                  final String metricsContent = senderEnv.getUrlContent(metricsUrl, null);
+                  final String metricsContent = env.getUrlContent(metricsUrl, null);
                   Assert.assertNotNull(
                       "Failed to fetch metrics from leader DataNode at " + metricsUrl,
                       metricsContent);
@@ -480,8 +494,8 @@ private List<TRegionInfo> showTableDataRegionLeaders(
     return result;
   }
 
-  private DataNodeWrapper findDataNodeWrapperByPort(final int port) {
-    for (final DataNodeWrapper dataNodeWrapper : senderEnv.getDataNodeWrapperList()) {
+  private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final int port) {
+    for (final DataNodeWrapper dataNodeWrapper : env.getDataNodeWrapperList()) {
       if (dataNodeWrapper.getPort() == port) {
         return dataNodeWrapper;
       }