[IOTDB-5968] Pipe: pipe task does not work properly after cluster reboot (apache#10046)

SteveYurongSu · web-flow · commit 5d78f1a915f3 · 2023-06-04T01:23:14.000+08:00
* fix: dead lock issue found in NodeManager.java

* fix: dead lock issue found in SimpleProgressIndex.java

* fix: logic error of 'pipe task start' found in PipeTaskAgent.java , which may cause data consistency issues

* fix: NPE error from ClusterSyncInfoFetcher.java

* improvement: reduce some exception log print

* improvement: introduce some log print for progress index reporting and recovering
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -206,8 +206,10 @@ private void setCQConfig(ConfigurationResp dataSet) {
   }
 
   private TRuntimeConfiguration getRuntimeConfiguration() {
-    getPipeManager().getPipePluginCoordinator().lock();
+    // getPipeTaskCoordinator.lock() should be called outside the getPipePluginCoordinator().lock()
+    // to avoid deadlock
     getPipeManager().getPipeTaskCoordinator().lock();
+    getPipeManager().getPipePluginCoordinator().lock();
     getTriggerManager().getTriggerInfo().acquireTriggerTableLock();
     getUDFManager().getUdfInfo().acquireUDFTableLock();
 
@@ -226,8 +228,11 @@ private TRuntimeConfiguration getRuntimeConfiguration() {
     } finally {
       getTriggerManager().getTriggerInfo().releaseTriggerTableLock();
       getUDFManager().getUdfInfo().releaseUDFTableLock();
-      getPipeManager().getPipeTaskCoordinator().unlock();
       getPipeManager().getPipePluginCoordinator().unlock();
+      // getPipeTaskCoordinator.unlock() should be called outside the
+      // getPipePluginCoordinator().unlock()
+      // to avoid deadlock
+      getPipeManager().getPipeTaskCoordinator().unlock();
     }
   }
 
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -223,22 +223,23 @@ public TSStatus handleLeaderChange(PipeHandleLeaderChangePlan plan) {
                                   dataRegionGroupId,
                                   new PipeTaskMeta(
                                       new MinimumProgressIndex(), newDataRegionLeader));
-                            } else {
-                              LOGGER.warn(
-                                  "The pipe task meta does not contain the data region group {} or the data region group has already been removed",
-                                  dataRegionGroupId);
                             }
+                            // else:
+                            // "The pipe task meta does not contain the data region group {} or
+                            // the data region group has already been removed"
                           }
                         }));
     return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
   }
 
   public TSStatus handleMetaChanges(PipeHandleMetaChangePlan plan) {
+    LOGGER.info("Handling pipe meta changes ...");
     pipeMetaKeeper.clear();
     plan.getPipeMetaList()
         .forEach(
             pipeMeta -> {
               pipeMetaKeeper.addPipeMeta(pipeMeta.getStaticMeta().getPipeName(), pipeMeta);
+              LOGGER.info("Recording pipe meta: {}", pipeMeta);
             });
     return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
   }
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
@@ -100,10 +100,10 @@ protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
   }
 
   @Override
-  protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+  protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
     LOGGER.info("PipeHandleLeaderChangeProcedure: executeFromHandleOnDataNodes");
 
-    pushPipeMetaToDataNodes(env);
+    pushPipeMetaToDataNodesIgnoreException(env);
   }
 
   @Override
@@ -142,10 +142,10 @@ protected void rollbackFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
   }
 
   @Override
-  protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+  protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
     LOGGER.info("PipeHandleLeaderChangeProcedure: rollbackFromCreateOnDataNodes");
 
-    pushPipeMetaToDataNodes(env);
+    pushPipeMetaToDataNodesIgnoreException(env);
   }
 
   @Override
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
@@ -138,9 +138,20 @@ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
             .getValue()
             .getProgressIndex()
             .isAfter(runtimeMetaFromDataNode.getProgressIndex())) {
-          runtimeMetaOnConfigNode
-              .getValue()
-              .updateProgressIndex(runtimeMetaFromDataNode.getProgressIndex());
+          LOGGER.info(
+              "Updating progress index for (pipe name: {}, consensus group id: {}) ... Progress index on config node: {}, progress index from data node: {}",
+              pipeMetaOnConfigNode.getStaticMeta().getPipeName(),
+              runtimeMetaOnConfigNode.getKey(),
+              runtimeMetaOnConfigNode.getValue().getProgressIndex(),
+              runtimeMetaFromDataNode.getProgressIndex());
+          LOGGER.info(
+              "Progress index for (pipe name: {}, consensus group id: {}) is updated to {}",
+              pipeMetaOnConfigNode.getStaticMeta().getPipeName(),
+              runtimeMetaOnConfigNode.getKey(),
+              runtimeMetaOnConfigNode
+                  .getValue()
+                  .updateProgressIndex(runtimeMetaFromDataNode.getProgressIndex()));
+
           needWriteConsensusOnConfigNodes = true;
         }
 
@@ -149,6 +160,7 @@ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
         pipeTaskMetaOnConfigNode.clearExceptionMessages();
         for (final PipeRuntimeException exception :
             runtimeMetaFromDataNode.getExceptionMessages()) {
+
           pipeTaskMetaOnConfigNode.trackExceptionMessage(exception);
 
           if (exception instanceof PipeRuntimeCriticalException) {
@@ -159,6 +171,7 @@ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
                 .get()
                 .equals(PipeStatus.STOPPED)) {
               pipeMetaOnConfigNode.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
+
               needWriteConsensusOnConfigNodes = true;
               needPushPipeMetaToDataNodes = true;
 
@@ -181,6 +194,7 @@ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
                       .forEach(
                           status -> {
                             status.set(PipeStatus.STOPPED);
+
                             needWriteConsensusOnConfigNodes = true;
                             needPushPipeMetaToDataNodes = true;
 
@@ -224,14 +238,14 @@ protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
   }
 
   @Override
-  protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+  protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
     LOGGER.info("PipeHandleMetaChangeProcedure: executeFromHandleOnDataNodes");
 
     if (!needPushPipeMetaToDataNodes) {
       return;
     }
 
-    pushPipeMetaToDataNodes(env);
+    pushPipeMetaToDataNodesIgnoreException(env);
   }
 
   @Override
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
@@ -66,10 +66,10 @@ protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
   }
 
   @Override
-  protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+  protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
     LOGGER.info("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
 
-    pushPipeMetaToDataNodes(env);
+    pushPipeMetaToDataNodesIgnoreException(env);
   }
 
   @Override
@@ -94,10 +94,10 @@ protected void rollbackFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
   }
 
   @Override
-  protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+  protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
     LOGGER.info("PipeMetaSyncProcedure: rollbackFromOperateOnDataNodes");
 
-    pushPipeMetaToDataNodes(env);
+    // do nothing
   }
 
   @Override
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java
@@ -200,6 +200,14 @@ protected void pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IOExce
     }
   }
 
+  protected void pushPipeMetaToDataNodesIgnoreException(ConfigNodeProcedureEnv env) {
+    try {
+      pushPipeMetaToDataNodes(env);
+    } catch (Throwable throwable) {
+      LOGGER.info("Failed to push pipe meta list to data nodes, will retry later.", throwable);
+    }
+  }
+
   @Override
   public void serialize(DataOutputStream stream) throws IOException {
     super.serialize(stream);
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SimpleProgressIndex.java b/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SimpleProgressIndex.java
@@ -165,7 +165,7 @@ public ProgressIndex updateToMinimumIsAfterProgressIndex(ProgressIndex progressI
       // thatSimpleProgressIndex.memtableFlushOrderId
       return this;
     } finally {
-      lock.writeLock().lock();
+      lock.writeLock().unlock();
     }
   }
 
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java
@@ -52,8 +52,9 @@ public ProgressIndex getProgressIndex() {
     return progressIndex.get();
   }
 
-  public void updateProgressIndex(ProgressIndex updateIndex) {
-    progressIndex.updateAndGet(index -> index.updateToMinimumIsAfterProgressIndex(updateIndex));
+  public ProgressIndex updateProgressIndex(ProgressIndex updateIndex) {
+    return progressIndex.updateAndGet(
+        index -> index.updateToMinimumIsAfterProgressIndex(updateIndex));
   }
 
   public int getLeaderDataNodeId() {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/HeartbeatScheduler.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/HeartbeatScheduler.java
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
@@ -163,13 +163,19 @@ public static synchronized void launchPipeTaskAgent() {
       PipeAgent.task()
           .handlePipeMetaChanges(
               getAllPipeInfoResp.getAllPipeInfo().stream()
-                  .map(PipeMeta::deserialize)
+                  .map(
+                      byteBuffer -> {
+                        final PipeMeta pipeMeta = PipeMeta.deserialize(byteBuffer);
+                        LOGGER.info(
+                            "Pulled pipe meta from config node: {}, recovering ...", pipeMeta);
+                        return pipeMeta;
+                      })
                   .collect(Collectors.toList()));
-    } catch (Throwable throwable) {
+    } catch (Exception e) {
       LOGGER.info(
           "Failed to get pipe task meta from config node. Ignore the exception, "
               + "because config node may not be ready yet, and meta will be pushed by config node later.",
-          throwable);
+          e);
     }
   }
 }
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
@@ -49,6 +49,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 /**
@@ -94,11 +95,11 @@ public synchronized void handlePipeMetaChanges(List<PipeMeta> pipeMetaListFromCo
 
       // if pipe meta does not exist on data node, create a new pipe
       if (metaOnDataNode == null) {
-        createPipe(metaFromConfigNode);
-        if (metaFromConfigNode.getRuntimeMeta().getStatus().get() == PipeStatus.RUNNING) {
+        if (createPipe(metaFromConfigNode)) {
+          // if the status recorded in config node is RUNNING, start the pipe
           startPipe(pipeName, metaFromConfigNode.getStaticMeta().getCreationTime());
         }
-        // if the status is STOPPED or DROPPED, do nothing
+        // if the status recorded in config node is STOPPED or DROPPED, do nothing
         continue;
       }
 
@@ -109,8 +110,7 @@ public synchronized void handlePipeMetaChanges(List<PipeMeta> pipeMetaListFromCo
       // first check if pipe static meta has changed, if so, drop the pipe and create a new one
       if (!staticMetaOnDataNode.equals(staticMetaFromConfigNode)) {
         dropPipe(pipeName);
-        createPipe(metaFromConfigNode);
-        if (metaFromConfigNode.getRuntimeMeta().getStatus().get() == PipeStatus.RUNNING) {
+        if (createPipe(metaFromConfigNode)) {
           startPipe(pipeName, metaFromConfigNode.getStaticMeta().getCreationTime());
         }
         // if the status is STOPPED or DROPPED, do nothing
@@ -245,9 +245,17 @@ public synchronized void dropAllPipeTasks() {
 
   ////////////////////////// Manage by Pipe Name //////////////////////////
 
-  private void createPipe(PipeMeta pipeMeta) {
-    final String pipeName = pipeMeta.getStaticMeta().getPipeName();
-    final long creationTime = pipeMeta.getStaticMeta().getCreationTime();
+  /**
+   * Create a new pipe. If the pipe already exists, do nothing and return false. Otherwise, create
+   * the pipe and return true.
+   *
+   * @param pipeMetaFromConfigNode pipe meta from config node
+   * @return true if the pipe is created successfully and should be started, false if the pipe
+   *     already exists or is created but should not be started
+   */
+  private boolean createPipe(PipeMeta pipeMetaFromConfigNode) {
+    final String pipeName = pipeMetaFromConfigNode.getStaticMeta().getPipeName();
+    final long creationTime = pipeMetaFromConfigNode.getStaticMeta().getCreationTime();
 
     final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName);
     if (existedPipeMeta != null) {
@@ -260,7 +268,7 @@ private void createPipe(PipeMeta pipeMeta) {
                 pipeName,
                 creationTime,
                 existedPipeMeta.getRuntimeMeta().getStatus().get().name());
-            return;
+            return false;
           case DROPPED:
             LOGGER.info(
                 "Pipe {} (creation time = {}) has already been dropped, but the pipe task meta has not been cleaned up. "
@@ -284,16 +292,25 @@ private void createPipe(PipeMeta pipeMeta) {
     }
 
     // create pipe tasks and trigger create() method for each pipe task
-    final Map<TConsensusGroupId, PipeTask> pipeTasks = new PipeBuilder(pipeMeta).build();
+    final Map<TConsensusGroupId, PipeTask> pipeTasks =
+        new PipeBuilder(pipeMetaFromConfigNode).build();
     for (PipeTask pipeTask : pipeTasks.values()) {
       pipeTask.create();
     }
-    pipeTaskManager.addPipeTasks(pipeMeta.getStaticMeta(), pipeTasks);
+    pipeTaskManager.addPipeTasks(pipeMetaFromConfigNode.getStaticMeta(), pipeTasks);
+
+    // No matter the pipe status from config node is RUNNING or STOPPED, we always set the status
+    // of pipe meta to STOPPED when it is created. The STOPPED status should always be the initial
+    // status of a pipe, which makes the status transition logic simpler.
+    final AtomicReference<PipeStatus> pipeStatusFromConfigNode =
+        pipeMetaFromConfigNode.getRuntimeMeta().getStatus();
+    final boolean needToStartPipe = pipeStatusFromConfigNode.get() == PipeStatus.RUNNING;
+    pipeStatusFromConfigNode.set(PipeStatus.STOPPED);
+
+    pipeMetaKeeper.addPipeMeta(pipeName, pipeMetaFromConfigNode);
 
-    // add pipe meta to pipe meta keeper
-    // note that we do not need to set the status of pipe meta here, because the status of pipe meta
-    // is already set to STOPPED when it is created
-    pipeMetaKeeper.addPipeMeta(pipeName, pipeMeta);
+    // If the pipe status from config node is RUNNING, we will start the pipe later.
+    return needToStartPipe;
   }
 
   private void dropPipe(String pipeName, long creationTime) {
@@ -551,14 +568,16 @@ private void stopPipeTask(TConsensusGroupId dataRegionGroupId, PipeStaticMeta pi
 
   public synchronized void collectPipeMetaList(THeartbeatReq req, THeartbeatResp resp)
       throws TException {
-    if (!req.isNeedPipeMetaList()) {
+    // do nothing if data node is removing or removed, or request does not need pipe meta list
+    if (PipeAgent.runtime().isShutdown() || !req.isNeedPipeMetaList()) {
       return;
     }
 
     final List<ByteBuffer> pipeMetaBinaryList = new ArrayList<>();
     try {
       for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
         pipeMetaBinaryList.add(pipeMeta.serialize());
+        LOGGER.info("Reporting pipe meta: {}", pipeMeta);
       }
     } catch (IOException e) {
       throw new TException(e);
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochManager.java b/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochManager.java
@@ -46,15 +46,13 @@ public PipeRealtimeCollectEvent bindPipeTsFileInsertionEvent(
 
     // this would not happen, but just in case
     if (!filePath2Epoch.containsKey(filePath)) {
-      LOGGER.warn(
-          String.format("PipeEngine: can not find TsFileEpoch for TsFile %s, create it", filePath));
+      LOGGER.info(
+          String.format("Pipe: can not find TsFileEpoch for TsFile %s, creating it", filePath));
       filePath2Epoch.put(filePath, new TsFileEpoch(filePath));
     }
 
     return new PipeRealtimeCollectEvent(
         event,
-        // TODO: we have to make sure that the TsFileInsertionEvent is the last event of the
-        // TsFileEpoch's life cycle
         filePath2Epoch.remove(filePath),
         resource.getDevices().stream()
             .collect(Collectors.toMap(device -> device, device -> EMPTY_MEASUREMENT_ARRAY)),
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/common/ClusterSyncInfoFetcher.java b/server/src/main/java/org/apache/iotdb/db/sync/common/ClusterSyncInfoFetcher.java

Original file line number	Diff line number	Diff line change
`@@ -100,10 +100,10 @@ protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {`
`100`	`100`	`}`
`101`	`101`
`102`	`102`	`@Override`
`103`		`- protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {`
	`103`	`+ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {`
`104`	`104`	`LOGGER.info("PipeHandleLeaderChangeProcedure: executeFromHandleOnDataNodes");`
`105`	`105`
`106`		`- pushPipeMetaToDataNodes(env);`
	`106`	`+ pushPipeMetaToDataNodesIgnoreException(env);`
`107`	`107`	`}`
`108`	`108`
`109`	`109`	`@Override`
`@@ -142,10 +142,10 @@ protected void rollbackFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)`
`142`	`142`	`}`
`143`	`143`
`144`	`144`	`@Override`
`145`		`- protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {`
	`145`	`+ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {`
`146`	`146`	`LOGGER.info("PipeHandleLeaderChangeProcedure: rollbackFromCreateOnDataNodes");`
`147`	`147`
`148`		`- pushPipeMetaToDataNodes(env);`
	`148`	`+ pushPipeMetaToDataNodesIgnoreException(env);`
`149`	`149`	`}`
`150`	`150`
`151`	`151`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -66,10 +66,10 @@ protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`@Override`
`69`		`- protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {`
	`69`	`+ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {`
`70`	`70`	`LOGGER.info("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");`
`71`	`71`
`72`		`- pushPipeMetaToDataNodes(env);`
	`72`	`+ pushPipeMetaToDataNodesIgnoreException(env);`
`73`	`73`	`}`
`74`	`74`
`75`	`75`	`@Override`
`@@ -94,10 +94,10 @@ protected void rollbackFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)`
`94`	`94`	`}`
`95`	`95`
`96`	`96`	`@Override`
`97`		`- protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {`
	`97`	`+ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {`
`98`	`98`	`LOGGER.info("PipeMetaSyncProcedure: rollbackFromOperateOnDataNodes");`
`99`	`99`
`100`		`- pushPipeMetaToDataNodes(env);`
	`100`	`+ // do nothing`
`101`	`101`	`}`
`102`	`102`
`103`	`103`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -200,6 +200,14 @@ protected void pushPipeMetaToDataNodes(ConfigNodeProcedureEnv env) throws IOExce`
`200`	`200`	`}`
`201`	`201`	`}`
`202`	`202`
	`203`	`+ protected void pushPipeMetaToDataNodesIgnoreException(ConfigNodeProcedureEnv env) {`
	`204`	`+ try {`
	`205`	`+ pushPipeMetaToDataNodes(env);`
	`206`	`+ } catch (Throwable throwable) {`
	`207`	`+ LOGGER.info("Failed to push pipe meta list to data nodes, will retry later.", throwable);`
	`208`	`+ }`
	`209`	`+ }`
	`210`	`+`
`203`	`211`	`@Override`
`204`	`212`	`public void serialize(DataOutputStream stream) throws IOException {`
`205`	`213`	`super.serialize(stream);`
Original file line number	Diff line number	Diff line change
`@@ -165,7 +165,7 @@ public ProgressIndex updateToMinimumIsAfterProgressIndex(ProgressIndex progressI`
`165`	`165`	`// thatSimpleProgressIndex.memtableFlushOrderId`
`166`	`166`	`return this;`
`167`	`167`	`} finally {`
`168`		`- lock.writeLock().lock();`
	`168`	`+ lock.writeLock().unlock();`
`169`	`169`	`}`
`170`	`170`	`}`
`171`	`171`
Original file line number	Diff line number	Diff line change
`@@ -52,8 +52,9 @@ public ProgressIndex getProgressIndex() {`
`52`	`52`	`return progressIndex.get();`
`53`	`53`	`}`
`54`	`54`
`55`		`- public void updateProgressIndex(ProgressIndex updateIndex) {`
`56`		`- progressIndex.updateAndGet(index -> index.updateToMinimumIsAfterProgressIndex(updateIndex));`
	`55`	`+ public ProgressIndex updateProgressIndex(ProgressIndex updateIndex) {`
	`56`	`+ return progressIndex.updateAndGet(`
	`57`	`+ index -> index.updateToMinimumIsAfterProgressIndex(updateIndex));`
`57`	`58`	`}`
`58`	`59`
`59`	`60`	`public int getLeaderDataNodeId() {`