Fixed the procedure has_more_state bug & Fixed the IT for encoding & compressor (#16729)

* IT-fix

* fix

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java

@@ -81,27 +81,35 @@ public void testPureSchemaInclusion() throws Exception {
               // banned
               "create timeSeries root.ln.wf01.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
               "ALTER timeSeries root.ln.wf01.wt01.status ADD TAGS tag3=v3",
-              "ALTER timeSeries root.ln.wf01.wt01.status ADD ATTRIBUTES attr4=v4",
-              "ALTER timeSeries root.** set compressor=ZSTD"),
+              "ALTER timeSeries root.ln.wf01.wt01.status ADD ATTRIBUTES attr4=v4"),
           null)) {
-        return;
+          return;
       }
 
       TestUtils.assertDataEventuallyOnEnv(
           receiverEnv,
           "show timeseries root.ln.**",
           "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
           Collections.singleton(
-              "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,ZSTD,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,"));
+              "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,LZ4,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,"));
 
       if (!TestUtils.tryExecuteNonQueriesWithRetry(
           senderEnv,
-          Arrays.asList(
-              "insert into root.ln.wf01.wt01(time, status) values(now(), false)", "flush"),
+              Arrays.asList(
+                      "ALTER timeSeries root.** set STORAGE_PROPERTIES compressor=ZSTD",
+                      "insert into root.ln.wf01.wt01(time, status) values(now(), false)",
+                      "flush"),
           null)) {
         return;
       }
 
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "show timeseries root.ln.**",
+          "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
+          Collections.singleton(
+              "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,ZSTD,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,"));
+
       TestUtils.assertDataAlwaysOnEnv(
           receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet());
     }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/AlterEncodingCompressorProcedure.java

@@ -125,7 +125,9 @@ protected Flow executeFromState(
                 SerializeUtils.deserializeCompressorNullable(compressor),
                 requestMessage);
           }
-          alterEncodingCompressorInSchemaRegion(env);
+          if (!alterEncodingCompressorInSchemaRegion(env)) {
+            return Flow.NO_MORE_STATE;
+          }
           break;
         case CLEAR_CACHE:
           LOGGER.info("Invalidate cache of timeSeries {}", requestMessage);
@@ -145,7 +147,7 @@ protected Flow executeFromState(
     }
   }
 
-  private void alterEncodingCompressorInSchemaRegion(final ConfigNodeProcedureEnv env) {
+  private boolean alterEncodingCompressorInSchemaRegion(final ConfigNodeProcedureEnv env) {
     final Map<TConsensusGroupId, TRegionReplicaSet> relatedSchemaRegionGroup =
         env.getConfigManager().getRelatedSchemaRegionGroup(patternTree, mayAlterAudit);
 
@@ -159,7 +161,7 @@ private void alterEncodingCompressorInSchemaRegion(final ConfigNodeProcedureEnv
                         .collect(Collectors.toList()),
                     false)));
       }
-      return;
+      return false;
     }
 
     final DataNodeTSStatusTaskExecutor<TAlterEncodingCompressorReq> alterEncodingCompressorTask =
@@ -222,6 +224,7 @@ protected void onAllReplicasetFailure(
         };
     alterEncodingCompressorTask.execute();
     setNextState(AlterEncodingCompressorState.CLEAR_CACHE);
+    return true;
   }
 
   private void collectPayload4Pipe(final ConfigNodeProcedureEnv env) {