Skip to content

Commit e3b9b36

Browse files
authored
[To dev/1.3] Pipe: Set placeholder for required pipeMetaList & Prevented the empty progress report when shutdown (#16388) (#16396)
* fix * refactor
1 parent 7462cbe commit e3b9b36

5 files changed

Lines changed: 10 additions & 6 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ public void stopPipeHeartbeat() {
104104

105105
public void parseHeartbeat(
106106
final int dataNodeId,
107-
/* @Nullable */ final List<ByteBuffer> pipeMetaByteBufferListFromDataNode,
107+
final List<ByteBuffer> pipeMetaByteBufferListFromDataNode,
108108
/* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
109109
/* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
110110
/* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent) {

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,11 +35,11 @@ public class PipeHeartbeat {
3535
private final Map<PipeStaticMeta, Double> remainingTimeMap = new HashMap<>();
3636

3737
public PipeHeartbeat(
38-
/* @Nullable */ final List<ByteBuffer> pipeMetaByteBufferListFromAgent,
38+
final List<ByteBuffer> pipeMetaByteBufferListFromAgent,
3939
/* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
4040
/* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
4141
/* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent) {
42-
// Pipe meta may be null for nodes shutting down, return empty heartbeat
42+
// Shall not reach here, just in case
4343
if (Objects.isNull(pipeMetaByteBufferListFromAgent)) {
4444
return;
4545
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.slf4j.Logger;
3737
import org.slf4j.LoggerFactory;
3838

39+
import java.util.ArrayList;
3940
import java.util.Map;
4041
import java.util.concurrent.Future;
4142
import java.util.concurrent.ScheduledExecutorService;
@@ -118,7 +119,7 @@ private synchronized void heartbeat() {
118119

119120
// config node heartbeat
120121
try {
121-
final TPipeHeartbeatResp configNodeResp = new TPipeHeartbeatResp();
122+
final TPipeHeartbeatResp configNodeResp = new TPipeHeartbeatResp(new ArrayList<>());
122123
PipeConfigNodeAgent.task().collectPipeMetaList(request, configNodeResp);
123124
pipeHeartbeatParser.parseHeartbeat(
124125
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -632,8 +632,11 @@ public void persistAllProgressIndex() {
632632
try (final ConfigNodeClient configNodeClient =
633633
ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
634634
// Send request to some API server
635-
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp();
635+
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>());
636636
collectPipeMetaList(new TPipeHeartbeatReq(Long.MIN_VALUE), resp);
637+
if (resp.getPipeMetaList().isEmpty()) {
638+
return;
639+
}
637640
final TSStatus result =
638641
configNodeClient.pushHeartbeat(
639642
IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1374,7 +1374,7 @@ public TPushConsumerGroupMetaResp pushSingleConsumerGroupMeta(
13741374

13751375
@Override
13761376
public TPipeHeartbeatResp pipeHeartbeat(TPipeHeartbeatReq req) throws TException {
1377-
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp();
1377+
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>());
13781378
PipeDataNodeAgent.task().collectPipeMetaList(req, resp);
13791379
return resp;
13801380
}

0 commit comments

Comments
 (0)