You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/09/12 13:22:26 UTC
[iotdb] branch 1.2-fix-pipe-npe created (now df11c603cf5)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch 1.2-fix-pipe-npe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at df11c603cf5 Pipe: fix NPE when HybridProgressIndex.updateToMinimumIsAfterProgressIndex after system reboot (DR: SimpleConsensus) (#11126)
This branch includes the following new commits:
new df11c603cf5 Pipe: fix NPE when HybridProgressIndex.updateToMinimumIsAfterProgressIndex after system reboot (DR: SimpleConsensus) (#11126)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: Pipe: fix NPE when HybridProgressIndex.updateToMinimumIsAfterProgressIndex after system reboot (DR: SimpleConsensus) (#11126)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch 1.2-fix-pipe-npe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit df11c603cf509fa93448faa213cb6459961321e8
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Sep 12 20:33:12 2023 +0800
Pipe: fix NPE when HybridProgressIndex.updateToMinimumIsAfterProgressIndex after system reboot (DR: SimpleConsensus) (#11126)
(cherry picked from commit b49d0fb96f2d4192abb734dd83883dd283ae1fb2)
---
.../confignode/persistence/pipe/PipeTaskInfo.java | 2 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 2 +-
.../request/ConfigPhysicalPlanSerDeTest.java | 6 +++---
.../consensus/response/pipe/PipeTableRespTest.java | 6 +++---
.../iotdb/confignode/persistence/PipeInfoTest.java | 4 ++--
.../apache/iotdb/db/pipe/event/EnrichedEvent.java | 5 ++++-
.../event/common/heartbeat/PipeHeartbeatEvent.java | 2 +-
.../tablet/PipeInsertNodeTabletInsertionEvent.java | 3 ++-
.../common/tablet/PipeRawTabletInsertionEvent.java | 2 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 2 +-
.../iotdb/db/pipe/task/subtask/PipeSubtask.java | 21 ++++++++++++---------
.../dataregion/tsfile/TsFileResource.java | 2 +-
.../commons/consensus/index/ProgressIndex.java | 4 ++--
.../consensus/index/impl/HybridProgressIndex.java | 2 +-
.../consensus/index/impl/MinimumProgressIndex.java | 8 +++++---
.../iotdb/commons/pipe/PipeMetaDeSerTest.java | 2 +-
16 files changed, 41 insertions(+), 32 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 7ffeeed3d77..615f518221d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -331,7 +331,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
consensusGroupIdToTaskMetaMap.put(
dataRegionGroupId,
new PipeTaskMeta(
- new MinimumProgressIndex(), newDataRegionLeader));
+ MinimumProgressIndex.INSTANCE, newDataRegionLeader));
}
// else:
// "The pipe task meta does not contain the data region group {} or
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
index 470483c7ac3..44dc4360a89 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -115,7 +115,7 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
// Pipe only collect user's data, filter metric database here.
consensusGroupIdToTaskMetaMap.put(
regionGroupId,
- new PipeTaskMeta(new MinimumProgressIndex(), regionLeaderNodeId));
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, regionLeaderNodeId));
}
}
});
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index b09b65d5085..a7e842ea31f 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -1061,7 +1061,7 @@ public class ConfigPhysicalPlanSerDeTest {
extractorAttributes.put("extractor", "org.apache.iotdb.pipe.extractor.DefaultExtractor");
processorAttributes.put("processor", "org.apache.iotdb.pipe.processor.SDTFilterProcessor");
connectorAttributes.put("connector", "org.apache.iotdb.pipe.protocal.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(new MinimumProgressIndex(), 1);
+ PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
@@ -1175,11 +1175,11 @@ public class ConfigPhysicalPlanSerDeTest {
put(
new TConsensusGroupId(TConsensusGroupType.DataRegion, 456),
new PipeTaskMeta(
- new MinimumProgressIndex(), 987)); // TODO: replace with IoTConsensus
+ MinimumProgressIndex.INSTANCE, 987)); // TODO: replace with IoTConsensus
put(
new TConsensusGroupId(TConsensusGroupType.DataRegion, 123),
new PipeTaskMeta(
- new MinimumProgressIndex(), 789)); // TODO: replace with IoTConsensus
+ MinimumProgressIndex.INSTANCE, 789)); // TODO: replace with IoTConsensus
}
});
pipeMetaList.add(new PipeMeta(pipeStaticMeta, pipeRuntimeMeta));
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
index 83109eceb0a..97967f97da0 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
@@ -55,7 +55,7 @@ public class PipeTableRespTest {
connectorAttributes.put("host", "127.0.0.1");
connectorAttributes.put("port", "6667");
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(new MinimumProgressIndex(), 1);
+ PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
@@ -75,7 +75,7 @@ public class PipeTableRespTest {
connectorAttributes1.put("host", "127.0.0.1");
connectorAttributes1.put("port", "6667");
- PipeTaskMeta pipeTaskMeta1 = new PipeTaskMeta(new MinimumProgressIndex(), 1);
+ PipeTaskMeta pipeTaskMeta1 = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
Map<TConsensusGroupId, PipeTaskMeta> pipeTasks1 = new HashMap<>();
pipeTasks1.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta1);
PipeStaticMeta pipeStaticMeta1 =
@@ -95,7 +95,7 @@ public class PipeTableRespTest {
connectorAttributes2.put("host", "172.30.30.30");
connectorAttributes2.put("port", "6667");
- PipeTaskMeta pipeTaskMeta2 = new PipeTaskMeta(new MinimumProgressIndex(), 1);
+ PipeTaskMeta pipeTaskMeta2 = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
Map<TConsensusGroupId, PipeTaskMeta> pipeTasks2 = new HashMap<>();
pipeTasks2.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta2);
PipeStaticMeta pipeStaticMeta2 =
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
index d032c96c7d9..a99db2ef9d1 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
@@ -85,7 +85,7 @@ public class PipeInfoTest {
connectorAttributes.put("host", "127.0.0.1");
connectorAttributes.put("port", "6667");
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(new MinimumProgressIndex(), 1);
+ PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
@@ -120,7 +120,7 @@ public class PipeInfoTest {
extractorAttributes.put("extractor", "org.apache.iotdb.pipe.extractor.DefaultExtractor");
processorAttributes.put("processor", "org.apache.iotdb.pipe.processor.SDTFilterProcessor");
connectorAttributes.put("connector", "org.apache.iotdb.pipe.protocal.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(new MinimumProgressIndex(), 1);
+ PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
index abe58d38857..0d304b4f1a8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
@@ -127,7 +128,9 @@ public abstract class EnrichedEvent implements Event {
protected void reportProgress() {
if (pipeTaskMeta != null) {
- pipeTaskMeta.updateProgressIndex(getProgressIndex());
+ final ProgressIndex progressIndex = getProgressIndex();
+ pipeTaskMeta.updateProgressIndex(
+ progressIndex == null ? MinimumProgressIndex.INSTANCE : progressIndex);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index 15f57aa94cf..ef5a7657b03 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -83,7 +83,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
@Override
public ProgressIndex getProgressIndex() {
- return new MinimumProgressIndex();
+ return MinimumProgressIndex.INSTANCE;
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
index 18aa4d721ec..9bb99a21bb2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.event.common.tablet;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
@@ -123,7 +124,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
@Override
public ProgressIndex getProgressIndex() {
- return progressIndex;
+ return progressIndex == null ? MinimumProgressIndex.INSTANCE : progressIndex;
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
index bd36d7eba56..a3f1925c7d5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -94,7 +94,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
@Override
public ProgressIndex getProgressIndex() {
- return sourceEvent != null ? sourceEvent.getProgressIndex() : new MinimumProgressIndex();
+ return sourceEvent != null ? sourceEvent.getProgressIndex() : MinimumProgressIndex.INSTANCE;
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 4c7d7f6fb0e..83642dd7417 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -156,7 +156,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
String.format(
"Interrupted when waiting for closing TsFile %s.", resource.getTsFilePath()));
Thread.currentThread().interrupt();
- return new MinimumProgressIndex();
+ return MinimumProgressIndex.INSTANCE;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
index ed780c89fbe..15d81348bb3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
@@ -72,17 +72,20 @@ public abstract class PipeSubtask
public Boolean call() throws Exception {
boolean hasAtLeastOneEventProcessed = false;
- // If the scheduler allows to schedule, then try to consume an event
- while (subtaskScheduler.schedule()) {
- // If the event is consumed successfully, then continue to consume the next event
- // otherwise, stop consuming
- if (!executeOnce()) {
- break;
+ try {
+ // If the scheduler allows to schedule, then try to consume an event
+ while (subtaskScheduler.schedule()) {
+ // If the event is consumed successfully, then continue to consume the next event
+ // otherwise, stop consuming
+ if (!executeOnce()) {
+ break;
+ }
+ hasAtLeastOneEventProcessed = true;
}
- hasAtLeastOneEventProcessed = true;
+ } finally {
+ // Reset the scheduler to make sure that the scheduler can schedule again
+ subtaskScheduler.reset();
}
- // Reset the scheduler to make sure that the scheduler can schedule again
- subtaskScheduler.reset();
return hasAtLeastOneEventProcessed;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index d31067b5a47..bcc4ff7664e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -1189,6 +1189,6 @@ public class TsFileResource {
}
public ProgressIndex getMaxProgressIndex() {
- return maxProgressIndex == null ? new MinimumProgressIndex() : maxProgressIndex;
+ return maxProgressIndex == null ? MinimumProgressIndex.INSTANCE : maxProgressIndex;
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java
index 4f2d641db57..0c789716f29 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java
@@ -109,10 +109,10 @@ public interface ProgressIndex {
static ProgressIndex blendProgressIndex(
ProgressIndex progressIndex1, ProgressIndex progressIndex2) {
if (progressIndex1 == null && progressIndex2 == null) {
- return new MinimumProgressIndex();
+ return MinimumProgressIndex.INSTANCE;
}
if (progressIndex1 == null || progressIndex1 instanceof MinimumProgressIndex) {
- return progressIndex2 == null ? new MinimumProgressIndex() : progressIndex2;
+ return progressIndex2 == null ? MinimumProgressIndex.INSTANCE : progressIndex2;
}
if (progressIndex2 == null || progressIndex2 instanceof MinimumProgressIndex) {
return progressIndex1; // progressIndex1 is not null
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java
index dcd701cfbe6..718970547cd 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java
@@ -159,7 +159,7 @@ public class HybridProgressIndex implements ProgressIndex {
public ProgressIndex updateToMinimumIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
try {
- if (progressIndex instanceof MinimumProgressIndex) {
+ if (progressIndex == null || progressIndex instanceof MinimumProgressIndex) {
return this;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java
index 6b9c37ec5e1..a5b587c6669 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java
@@ -29,7 +29,9 @@ import java.nio.ByteBuffer;
public class MinimumProgressIndex implements ProgressIndex {
- public MinimumProgressIndex() {}
+ public static final MinimumProgressIndex INSTANCE = new MinimumProgressIndex();
+
+ private MinimumProgressIndex() {}
@Override
public void serialize(ByteBuffer byteBuffer) {
@@ -78,11 +80,11 @@ public class MinimumProgressIndex implements ProgressIndex {
}
public static MinimumProgressIndex deserializeFrom(ByteBuffer byteBuffer) {
- return new MinimumProgressIndex();
+ return INSTANCE;
}
public static MinimumProgressIndex deserializeFrom(InputStream stream) {
- return new MinimumProgressIndex();
+ return INSTANCE;
}
@Override
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
index 56b25867e21..9eee8396d03 100644
--- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
@@ -76,7 +76,7 @@ public class PipeMetaDeSerTest {
{
put(
new TConsensusGroupId(TConsensusGroupType.DataRegion, 123),
- new PipeTaskMeta(new MinimumProgressIndex(), 987));
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 987));
put(
new TConsensusGroupId(TConsensusGroupType.DataRegion, 234),
new PipeTaskMeta(new IoTProgressIndex(1, 2L), 789));