Fix ser

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java

@@ -552,18 +552,15 @@ public boolean equals(final Object o) {
     }
     final CreatePipeProcedureV2 that = (CreatePipeProcedureV2) o;
     return this.createPipeRequest.getPipeName().equals(that.createPipeRequest.getPipeName())
-        && this.createPipeRequest
-            .getExtractorAttributes()
-            .toString()
-            .equals(that.createPipeRequest.getExtractorAttributes().toString())
-        && this.createPipeRequest
-            .getProcessorAttributes()
-            .toString()
-            .equals(that.createPipeRequest.getProcessorAttributes().toString())
-        && this.createPipeRequest
-            .getConnectorAttributes()
-            .toString()
-            .equals(that.createPipeRequest.getConnectorAttributes().toString());
+        && Objects.equals(
+            this.createPipeRequest.getExtractorAttributes(),
+            that.createPipeRequest.getExtractorAttributes())
+        && Objects.equals(
+            this.createPipeRequest.getProcessorAttributes(),
+            that.createPipeRequest.getProcessorAttributes())
+        && Objects.equals(
+            this.createPipeRequest.getConnectorAttributes(),
+            that.createPipeRequest.getConnectorAttributes());
   }
 
   @Override

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java

@@ -47,6 +47,7 @@
 public class DropPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DropPipeProcedureV2.class);
+  private static final int SERIALIZATION_VERSION_MAGIC = 0x44505632;
 
   private String pipeName;
   private boolean isTableModel;
@@ -211,6 +212,8 @@ public void serialize(DataOutputStream stream) throws IOException {
     stream.writeShort(ProcedureType.DROP_PIPE_PROCEDURE_V2.getTypeCode());
     super.serialize(stream);
     ReadWriteIOUtils.write(pipeName, stream);
+    ReadWriteIOUtils.write(SERIALIZATION_VERSION_MAGIC, stream);
+    ReadWriteIOUtils.write(isTableModelSet, stream);
     if (isTableModelSet) {
       ReadWriteIOUtils.write(isTableModel, stream);
       if (pipeMetaToDrop == null) {
@@ -229,6 +232,23 @@ public void deserialize(ByteBuffer byteBuffer) {
     if (!byteBuffer.hasRemaining()) {
       return;
     }
+    if (byteBuffer.remaining() >= Integer.BYTES) {
+      final int position = byteBuffer.position();
+      if (ReadWriteIOUtils.readInt(byteBuffer) == SERIALIZATION_VERSION_MAGIC) {
+        if (!byteBuffer.hasRemaining()) {
+          return;
+        }
+        isTableModelSet = ReadWriteIOUtils.readBool(byteBuffer);
+        if (isTableModelSet) {
+          isTableModel = ReadWriteIOUtils.readBool(byteBuffer);
+          if (ReadWriteIOUtils.readBool(byteBuffer)) {
+            pipeMetaToDrop = PipeMeta.deserialize4Coordinator(byteBuffer);
+          }
+        }
+        return;
+      }
+      byteBuffer.position(position);
+    }
     isTableModel = ReadWriteIOUtils.readBool(byteBuffer);
     isTableModelSet = true;
     if (byteBuffer.hasRemaining() && ReadWriteIOUtils.readBool(byteBuffer)) {