Skip to content

Commit 505b18a

Browse files
CaideyipiCaideyipi
authored
fix(pipe): reuse safe local progress on alter (#18423) (#18444)
* fix(pipe): reuse safe local progress on alter * fix(pipe): collect latest progress before alter
1 parent 79c9459 commit 505b18a

8 files changed

Lines changed: 524 additions & 8 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,15 @@
4747
@Category({MultiClusterIT2AutoCreateSchema.class})
4848
public class IoTDBPipeAlterIT extends AbstractPipeDualAutoIT {
4949

50+
@Override
51+
protected void setupConfig() {
52+
super.setupConfig();
53+
senderEnv
54+
.getConfig()
55+
.getCommonConfig()
56+
.setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(600);
57+
}
58+
5059
@Test
5160
public void testBasicAlterPipe() throws Exception {
5261
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
@@ -579,4 +588,43 @@ public void testAlterPipeRealtime() {
579588
"count(timeseries),",
580589
Collections.singleton("1,"));
581590
}
591+
592+
@Test
593+
public void testAlterPipeDoesNotResendCommittedData() {
594+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
595+
596+
TestUtils.executeNonQueries(
597+
senderEnv,
598+
Arrays.asList("insert into root.db.d1(time, s1) values (1, 1), (2, 2)", "flush"),
599+
null);
600+
601+
TestUtils.executeNonQuery(
602+
senderEnv,
603+
String.format(
604+
"create pipe a2b with source ('source.realtime.mode'='stream') with sink ('node-urls'='%s', 'sink.batch.enable'='false')",
605+
receiverDataNode.getIpAndPortString()),
606+
null);
607+
608+
final Set<String> oldData = new HashSet<>(Arrays.asList("1,1.0,", "2,2.0,"));
609+
TestUtils.assertDataEventuallyOnEnv(
610+
receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", oldData);
611+
612+
TestUtils.executeNonQuery(
613+
receiverEnv, "delete from root.db.d1.s1 where time >= 1 and time <= 2", null);
614+
TestUtils.assertDataEventuallyOnEnv(
615+
receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet());
616+
617+
TestUtils.executeNonQuery(
618+
senderEnv, "alter pipe a2b modify sink ('sink.batch.enable'='true')", null);
619+
TestUtils.assertDataAlwaysOnEnv(
620+
receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet());
621+
622+
TestUtils.executeNonQueries(
623+
senderEnv, Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"), null);
624+
final Set<String> newData = Collections.singleton("3,3.0,");
625+
TestUtils.assertDataEventuallyOnEnv(
626+
receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData);
627+
TestUtils.assertDataAlwaysOnEnv(
628+
receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData);
629+
}
582630
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
2525
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
2626
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
27+
import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
2728
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
2829
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2930
import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -73,6 +74,7 @@
7374
import org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq;
7475
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
7576
import org.apache.iotdb.mpp.rpc.thrift.TNotifyRegionMigrationReq;
77+
import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatReq;
7678
import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaReq;
7779
import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
7880
import org.apache.iotdb.mpp.rpc.thrift.TPushMultiPipeMetaReq;
@@ -953,6 +955,37 @@ public List<TSStatus> dropSingleConsumerGroupOnDataNode(String consumerGroupName
953955
.collect(Collectors.toList());
954956
}
955957

958+
/**
959+
* Collect the current pipe metadata from the specified DataNodes before a metadata-changing
960+
* procedure. The caller can use the returned task progress to avoid basing a replacement pipe on
961+
* a stale ConfigNode heartbeat.
962+
*
963+
* <p>This is deliberately best effort. A DataNode that is unavailable cannot contribute a newer
964+
* checkpoint, but the alter procedure can still use the checkpoint already stored by ConfigNode.
965+
*/
966+
public Map<Integer, TPipeHeartbeatResp> collectPipeMetaFromDataNodes(
967+
final Set<Integer> dataNodeIds) {
968+
if (dataNodeIds.isEmpty()) {
969+
return Collections.emptyMap();
970+
}
971+
972+
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
973+
configManager.getNodeManager().getRegisteredDataNodeLocations().entrySet().stream()
974+
.filter(entry -> dataNodeIds.contains(entry.getKey()))
975+
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
976+
if (dataNodeLocationMap.isEmpty()) {
977+
return Collections.emptyMap();
978+
}
979+
980+
final DataNodeAsyncRequestContext<TPipeHeartbeatReq, TPipeHeartbeatResp> clientHandler =
981+
new DataNodeAsyncRequestContext<>(
982+
CnToDnAsyncRequestType.PIPE_HEARTBEAT,
983+
new TPipeHeartbeatReq(System.currentTimeMillis()),
984+
dataNodeLocationMap);
985+
sendRuntimeMetaRequest(clientHandler, true, getRuntimeMetaPushTimeoutInMs());
986+
return clientHandler.getResponseMap();
987+
}
988+
956989
private static Map<Integer, TPushPipeMetaResp> sendPipeMetaRequest(
957990
final DataNodeAsyncRequestContext<?, TPushPipeMetaResp> clientHandler,
958991
final long timeoutInMs,

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

Lines changed: 80 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,9 @@
2020
package org.apache.iotdb.confignode.procedure.impl.pipe.task;
2121

2222
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
23+
import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
2324
import org.apache.iotdb.common.rpc.thrift.TSStatus;
25+
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
2426
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
2527
import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
2628
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
@@ -33,6 +35,7 @@
3335
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
3436
import org.apache.iotdb.confignode.consensus.request.write.pipe.task.AlterPipePlanV2;
3537
import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager;
38+
import org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat.PipeHeartbeat;
3639
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
3740
import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
3841
import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
@@ -49,9 +52,12 @@
4952
import java.io.DataOutputStream;
5053
import java.io.IOException;
5154
import java.nio.ByteBuffer;
55+
import java.util.Collections;
5256
import java.util.HashMap;
57+
import java.util.HashSet;
5358
import java.util.Map;
5459
import java.util.Objects;
60+
import java.util.Set;
5561
import java.util.concurrent.ConcurrentHashMap;
5662
import java.util.concurrent.ConcurrentMap;
5763

@@ -138,6 +144,16 @@ public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
138144
new HashMap<>(alterPipeRequest.getProcessorAttributes()),
139145
new HashMap<>(alterPipeRequest.getConnectorAttributes()));
140146

147+
// The periodic heartbeat may not have reached ConfigNode immediately before this alter. Pull
148+
// the current leader checkpoints first so a leader migration does not make the replacement
149+
// task start from an older coordinator checkpoint.
150+
final Map<Integer, ProgressIndex> latestProgressIndexMap =
151+
PipeTaskAgent.isRealtimeOnlyPipe(currentPipeStaticMeta.getExtractorParameters())
152+
== PipeTaskAgent.isRealtimeOnlyPipe(updatedPipeStaticMeta.getExtractorParameters())
153+
? collectLatestProgressIndexes(
154+
env, currentPipeStaticMeta, currentConsensusGroupId2PipeTaskMeta)
155+
: Collections.emptyMap();
156+
141157
final ConcurrentMap<Integer, PipeTaskMeta> updatedConsensusGroupIdToTaskMetaMap =
142158
new ConcurrentHashMap<>();
143159

@@ -167,8 +183,7 @@ public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
167183
// then it will extract all existing data now, not existing data since the
168184
// original pipe was created
169185
// Similar for "pure realtime"
170-
updatedConsensusGroupIdToTaskMetaMap.put(
171-
regionGroupId.getId(),
186+
final PipeTaskMeta updatedPipeTaskMeta =
172187
new PipeTaskMeta(
173188
PipeTaskAgent.isRealtimeOnlyPipe(
174189
currentPipeStaticMeta.getExtractorParameters())
@@ -186,7 +201,14 @@ public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
186201
&& PipeTaskAgent.isRealtimeOnlyPipe(
187202
updatedPipeStaticMeta.getExtractorParameters()))
188203
? PipeTaskMeta.getRevertedLeader(regionLeaderNodeId)
189-
: regionLeaderNodeId));
204+
: regionLeaderNodeId);
205+
final ProgressIndex latestProgressIndex =
206+
latestProgressIndexMap.get(regionGroupId.getId());
207+
if (latestProgressIndex != null) {
208+
updatedPipeTaskMeta.updateProgressIndex(latestProgressIndex);
209+
}
210+
updatedConsensusGroupIdToTaskMetaMap.put(
211+
regionGroupId.getId(), updatedPipeTaskMeta);
190212
}
191213
});
192214

@@ -213,6 +235,61 @@ public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
213235
}
214236
}
215237

238+
private Map<Integer, ProgressIndex> collectLatestProgressIndexes(
239+
final ConfigNodeProcedureEnv env,
240+
final PipeStaticMeta pipeStaticMeta,
241+
final Map<Integer, PipeTaskMeta> taskMetaMap) {
242+
final Set<Integer> leaderNodeIds = new HashSet<>();
243+
final Set<Integer> registeredDataNodeIds =
244+
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations().keySet();
245+
taskMetaMap.forEach(
246+
(consensusGroupId, taskMeta) -> {
247+
// The ConfigRegion task is led by a ConfigNode, not by a DataNode.
248+
if (consensusGroupId != Integer.MIN_VALUE
249+
&& registeredDataNodeIds.contains(taskMeta.getLeaderNodeId())) {
250+
leaderNodeIds.add(taskMeta.getLeaderNodeId());
251+
}
252+
});
253+
254+
if (leaderNodeIds.isEmpty()) {
255+
return Collections.emptyMap();
256+
}
257+
258+
final Map<Integer, TPipeHeartbeatResp> responseMap =
259+
env.collectPipeMetaFromDataNodes(leaderNodeIds);
260+
final Map<Integer, ProgressIndex> latestProgressIndexMap = new HashMap<>();
261+
responseMap.forEach(
262+
(dataNodeId, response) -> {
263+
if (response == null || !response.isSetPipeMetaList()) {
264+
return;
265+
}
266+
267+
final PipeMeta pipeMetaFromDataNode =
268+
new PipeHeartbeat(response.getPipeMetaList(), null, null, null)
269+
.getPipeMeta(pipeStaticMeta);
270+
if (pipeMetaFromDataNode == null) {
271+
return;
272+
}
273+
274+
pipeMetaFromDataNode
275+
.getRuntimeMeta()
276+
.getConsensusGroupId2TaskMetaMap()
277+
.forEach(
278+
(consensusGroupId, taskMetaFromDataNode) -> {
279+
final PipeTaskMeta taskMetaFromCoordinator = taskMetaMap.get(consensusGroupId);
280+
if (taskMetaFromCoordinator == null
281+
|| taskMetaFromCoordinator.getLeaderNodeId() != dataNodeId) {
282+
return;
283+
}
284+
latestProgressIndexMap.merge(
285+
consensusGroupId,
286+
taskMetaFromDataNode.getProgressIndex(),
287+
ProgressIndex::updateToMinimumEqualOrIsAfterProgressIndex);
288+
});
289+
});
290+
return latestProgressIndexMap;
291+
}
292+
216293
@Override
217294
public void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) throws PipeException {
218295
LOGGER.info(

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

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,7 @@
101101
import java.util.concurrent.ThreadPoolExecutor;
102102
import java.util.concurrent.TimeUnit;
103103
import java.util.concurrent.atomic.AtomicBoolean;
104+
import java.util.function.BiPredicate;
104105
import java.util.function.Consumer;
105106
import java.util.stream.Collectors;
106107
import java.util.stream.StreamSupport;
@@ -198,6 +199,8 @@ public List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChangesInternal(
198199
return Collections.emptyList();
199200
}
200201

202+
carryOverLocalProgressIndexForAlter(pipeMetaListFromCoordinator);
203+
201204
final List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
202205
super.handlePipeMetaChangesInternal(pipeMetaListFromCoordinator);
203206

@@ -218,6 +221,86 @@ public List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChangesInternal(
218221
return exceptionMessages;
219222
}
220223

224+
/**
225+
* Carry the committed progress of an old local task into an altered task when it is safe to do
226+
* so. The old task is dropped before the new task is created, therefore this must run before
227+
* {@link PipeTaskAgent#handlePipeMetaChangesInternal(List)} starts applying the metadata list.
228+
*
229+
* <p>We deliberately only carry progress when the old and new task stay on this DataNode and
230+
* their realtime-only modes are unchanged. Mode changes have explicit progress semantics in the
231+
* ConfigNode metadata (for example, realtime-only to historical resets to {@code
232+
* MinimumProgressIndex}), and leader changes must use the coordinator checkpoint because the old
233+
* task is not local to the new leader.
234+
*/
235+
private void carryOverLocalProgressIndexForAlter(
236+
final List<PipeMeta> pipeMetaListFromCoordinator) {
237+
for (final PipeMeta droppedPipeMeta : pipeMetaListFromCoordinator) {
238+
if (droppedPipeMeta.getRuntimeMeta().getStatus().get() != PipeStatus.DROPPED) {
239+
continue;
240+
}
241+
242+
final PipeStaticMeta oldStaticMeta = droppedPipeMeta.getStaticMeta();
243+
final PipeMeta localOldPipeMeta = pipeMetaKeeper.getPipeMeta(oldStaticMeta.getPipeName());
244+
if (localOldPipeMeta == null) {
245+
continue;
246+
}
247+
248+
for (final PipeMeta updatedPipeMeta : pipeMetaListFromCoordinator) {
249+
if (updatedPipeMeta == droppedPipeMeta
250+
|| updatedPipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.DROPPED
251+
|| !oldStaticMeta.getPipeName().equals(updatedPipeMeta.getStaticMeta().getPipeName())) {
252+
continue;
253+
}
254+
255+
carryOverLocalProgressIndexForAlter(
256+
oldStaticMeta,
257+
localOldPipeMeta,
258+
updatedPipeMeta,
259+
CONFIG.getDataNodeId(),
260+
(staticMeta, consensusGroupId) ->
261+
pipeTaskManager.getPipeTask(staticMeta, consensusGroupId) != null);
262+
}
263+
}
264+
}
265+
266+
static void carryOverLocalProgressIndexForAlter(
267+
final PipeStaticMeta oldStaticMeta,
268+
final PipeMeta localOldPipeMeta,
269+
final PipeMeta updatedPipeMeta,
270+
final int localNodeId,
271+
final BiPredicate<PipeStaticMeta, Integer> localTaskExists) {
272+
final PipeStaticMeta updatedStaticMeta = updatedPipeMeta.getStaticMeta();
273+
274+
// A mode change has an explicit cutover/reset meaning in ConfigNode. In particular, a
275+
// realtime-only -> historical alter must retain MinimumProgressIndex to scan old files.
276+
if (PipeTaskAgent.isRealtimeOnlyPipe(oldStaticMeta.getExtractorParameters())
277+
!= PipeTaskAgent.isRealtimeOnlyPipe(updatedStaticMeta.getExtractorParameters())) {
278+
return;
279+
}
280+
281+
final Map<Integer, PipeTaskMeta> localTaskMetaMap =
282+
localOldPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
283+
final Map<Integer, PipeTaskMeta> updatedTaskMetaMap =
284+
updatedPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
285+
286+
for (final Map.Entry<Integer, PipeTaskMeta> entry : updatedTaskMetaMap.entrySet()) {
287+
final int consensusGroupId = entry.getKey();
288+
final PipeTaskMeta updatedTaskMeta = entry.getValue();
289+
final PipeTaskMeta localTaskMeta = localTaskMetaMap.get(consensusGroupId);
290+
291+
// Only the old task's actual leader owns an authoritative local checkpoint. Requiring the
292+
// new task to stay on the same node also avoids losing the checkpoint during leader change.
293+
if (localTaskMeta == null
294+
|| localTaskMeta.getLeaderNodeId() != localNodeId
295+
|| updatedTaskMeta.getLeaderNodeId() != localNodeId
296+
|| !localTaskExists.test(oldStaticMeta, consensusGroupId)) {
297+
continue;
298+
}
299+
300+
updatedTaskMeta.updateProgressIndex(localTaskMeta.getProgressIndex());
301+
}
302+
}
303+
221304
private Set<Integer> clearSchemaRegionListeningQueueIfNecessary(
222305
final List<PipeMeta> pipeMetaListFromCoordinator) throws IllegalPathException {
223306
final Map<Integer, Long> schemaRegionId2ListeningQueueNewFirstIndex = new HashMap<>();

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1163,6 +1163,24 @@ public TPushPipeMetaRespExceptionMessage handleDropPipe(final String pipeName) {
11631163
return PipeDataNodeAgent.task().handleDropPipe(pipeName);
11641164
}
11651165

1166+
@Override
1167+
public boolean handlePipeMetaChanges(
1168+
final List<ByteBuffer> pipeMetas,
1169+
final List<TPushPipeMetaRespExceptionMessage> exceptionMessages) {
1170+
final List<TPushPipeMetaRespExceptionMessage> exceptionMessagesFromAgent =
1171+
PipeDataNodeAgent.task()
1172+
.handlePipeMetaChanges(
1173+
pipeMetas.stream()
1174+
.map(PipeMeta::deserialize4TaskAgent)
1175+
.collect(Collectors.toList()));
1176+
// PipeTaskAgent returns null only when its timed write-lock acquisition fails.
1177+
if (exceptionMessagesFromAgent == null) {
1178+
return false;
1179+
}
1180+
exceptionMessages.addAll(exceptionMessagesFromAgent);
1181+
return true;
1182+
}
1183+
11661184
@Override
11671185
public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(final ByteBuffer pipeMeta) {
11681186
return PipeDataNodeAgent.task()

0 commit comments

Comments
 (0)