gr

Author: Caideyipi

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

@@ -58,6 +58,7 @@ public class SetTTLProcedure extends StateMachineProcedure<ConfigNodeProcedureEn
   private static final Logger LOGGER = LoggerFactory.getLogger(SetTTLProcedure.class);
   // Distinguishes no previous TTL from TTLCache.NULL_TTL, the explicit unset marker for rollback.
   private static final long TTL_NOT_EXIST = Long.MIN_VALUE;
+  private static final int ROLLBACK_STATE_BYTES = Byte.BYTES + Long.BYTES * 2;
 
   private SetTTLPlan plan;
   private long previousTTL = TTL_NOT_EXIST;
@@ -202,8 +203,10 @@ private void rollbackConfigNodeTTL(final ConfigNodeProcedureEnv env) throws Proc
   private void restoreTTLOnConfigNode(
       final ConfigNodeProcedureEnv env, final String[] pathPattern, final long ttl)
       throws ProcedureException {
+    // TTL_NOT_EXIST means the original ttl was absent; NULL_TTL asks the executor to unset it.
     final SetTTLPlan rollbackPlan =
         new SetTTLPlan(pathPattern, ttl == TTL_NOT_EXIST ? TTLCache.NULL_TTL : ttl);
+    // Database rollback restores the database path and db.** separately, so avoid auto-expansion.
     rollbackPlan.setDataBase(false);
     final TSStatus status = writeConfigNodePlan(env, rollbackPlan);
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -247,6 +250,9 @@ private void restoreTTLOnDataNodes(
     }
   }
 
+  /**
+   * Best-effort rollback: restore both sides, throw the earliest failure, and suppress later ones.
+   */
   @Override
   protected void rollbackState(final ConfigNodeProcedureEnv env, final SetTTLState setTTLState)
       throws IOException, InterruptedException, ProcedureException {
@@ -266,6 +272,8 @@ protected void rollbackState(final ConfigNodeProcedureEnv env, final SetTTLState
       LOGGER.error("Failed to rollback DataNode ttl cache.", e);
       if (rollbackFailure == null) {
         rollbackFailure = e;
+      } else {
+        rollbackFailure.addSuppressed(e);
       }
     }
     if (rollbackFailure != null) {
@@ -313,8 +321,9 @@ public void deserialize(ByteBuffer byteBuffer) {
       final int length = ReadWriteIOUtils.readInt(byteBuffer);
       final int position = byteBuffer.position();
       this.plan = (SetTTLPlan) ConfigPhysicalPlan.Factory.create(byteBuffer);
+      // The serialized plan buffer may include padding; skip to the actual payload end.
       byteBuffer.position(position + length);
-      if (byteBuffer.remaining() >= 17) {
+      if (byteBuffer.remaining() >= ROLLBACK_STATE_BYTES) {
         this.previousTTLStateCaptured = byteBuffer.get() != 0;
         this.previousTTL = byteBuffer.getLong();
         this.previousDatabaseWildcardTTL = byteBuffer.getLong();

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedureTest.java

@@ -31,12 +31,16 @@
 import org.apache.iotdb.confignode.manager.ConfigManager;
 import org.apache.iotdb.confignode.manager.TTLManager;
 import org.apache.iotdb.confignode.manager.node.NodeManager;
+import org.apache.iotdb.confignode.procedure.Procedure;
 import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.state.ProcedureState;
 import org.apache.iotdb.confignode.procedure.state.schema.SetTTLState;
 import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.tsfile.utils.PublicBAOS;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;
@@ -111,6 +115,25 @@ public void serializeDeserializeTestWithCapturedRollbackState() throws Exception
         deserializedProcedure, "root.db", 2000L, true, true, 500L, 600L, false);
   }
 
+  @Test
+  public void deserializeOldFormatWithoutRollbackStateTest() throws Exception {
+    final SetTTLPlan setTTLPlan =
+        new SetTTLPlan(Arrays.asList(new PartialPath("root.db").getNodes()), 2000L);
+    setTTLPlan.setDataBase(true);
+
+    final PublicBAOS byteArrayOutputStream = new PublicBAOS();
+    final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+    writeOldFormatProcedure(outputStream, setTTLPlan);
+
+    final ByteBuffer buffer =
+        ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+    final SetTTLProcedure deserializedProcedure =
+        (SetTTLProcedure) ProcedureFactory.getInstance().create(buffer);
+
+    assertSerializedProcedure(
+        deserializedProcedure, "root.db", 2000L, true, false, Long.MIN_VALUE, Long.MIN_VALUE, false);
+  }
+
   @Test
   public void setConfigNodeTTLShouldNotWriteBeforePreviousStateIsCaptured() throws Exception {
     final SetTTLPlan setTTLPlan =
@@ -229,6 +252,25 @@ private void assertRequest(
     Assert.assertEquals(isDataBase, req.isDataBase);
   }
 
+  private void writeOldFormatProcedure(final DataOutputStream stream, final SetTTLPlan plan)
+      throws IOException {
+    stream.writeShort(ProcedureType.SET_TTL_PROCEDURE.getTypeCode());
+    // Procedure fields.
+    stream.writeLong(Procedure.NO_PROC_ID);
+    stream.writeInt(ProcedureState.INITIALIZING.ordinal());
+    stream.writeLong(0L);
+    stream.writeLong(0L);
+    stream.writeLong(Procedure.NO_PROC_ID);
+    stream.writeLong(Procedure.NO_TIMEOUT);
+    stream.writeInt(-1); // no stack indexes
+    stream.write((byte) 0); // no exception
+    stream.writeInt(-1); // no result
+    stream.write((byte) 0); // no lock
+    // StateMachineProcedure fields.
+    stream.writeInt(0); // no states
+    ReadWriteIOUtils.write(plan.serializeToByteBuffer(), stream);
+  }
+
   private void assertSerializedProcedure(
       final SetTTLProcedure procedure,
       final String path,