Caideyipi commented on code in PR #18423:
URL: https://github.com/apache/iotdb/pull/18423#discussion_r3755662506
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -217,6 +221,88 @@ public List<TPushPipeMetaRespExceptionMessage>
handlePipeMetaChangesInternal(
return exceptionMessages;
}
+ /**
+ * Carry the committed progress of an old local task into an altered task
when it is safe to do
+ * so. The old task is dropped before the new task is created, therefore
this must run before
+ * {@link PipeTaskAgent#handlePipeMetaChangesInternal(List)} starts applying
the metadata list.
+ *
+ * <p>We deliberately only carry progress when the old and new task stay on
this DataNode and
+ * their realtime-only modes are unchanged. Mode changes have explicit
progress semantics in the
+ * ConfigNode metadata (for example, realtime-only to historical resets to
{@code
+ * MinimumProgressIndex}), and leader changes must use the coordinator
checkpoint because the old
+ * task is not local to the new leader.
Review Comment:
Implemented in 5fb019b8d2c. Before building the altered metadata, ConfigNode
now pulls a fresh pipe heartbeat from the DataNodes recorded as the current
task leaders and monotonically merges each leader-reported checkpoint into the
replacement task metadata. It ignores ConfigRegion and non-leader reports, and
keeps the pull best-effort. The merge is skipped when realtime-only mode
changes so the existing progress-reset semantics are preserved.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java:
##########
@@ -60,11 +81,9 @@ static TPushPipeMetaResp pushMultiPipeMeta(
}
}
} else if (req.isSetPipeMetas()) {
- for (final ByteBuffer pipeMeta : req.getPipeMetas()) {
- final TPushPipeMetaRespExceptionMessage message =
handler.handleSinglePipeMeta(pipeMeta);
- if (message != null) {
- exceptionMessages.add(message);
- }
+ if (!handler.handlePipeMetaChanges(req.getPipeMetas(),
exceptionMessages)) {
+ return new TPushPipeMetaResp()
+ .setStatus(new
TSStatus(TSStatusCode.PIPE_PUSH_META_TIMEOUT.getStatusCode()));
Review Comment:
Added the explanation in 5fb019b8d2c. PipeTaskAgent.handlePipeMetaChanges
returns null only when tryWriteLockWithTimeOutInMs fails. The RPC adapter maps
null to false, and PushMultiPipeMetaHelper maps false to
PIPE_PUSH_META_TIMEOUT. Per-pipe errors stay in exceptionMessages and are
returned as PIPE_PUSH_META_ERROR.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java:
##########
@@ -46,6 +46,27 @@ interface Handler {
TPushPipeMetaRespExceptionMessage handleDropPipe(String pipeName) throws
Exception;
TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(ByteBuffer
pipeMeta) throws Exception;
+
+ /**
+ * Handles all pipe metadata in one agent invocation. Alter pipe sends the
old dropped metadata
+ * and the new metadata together, so they must be visible to the agent at
the same time when it
+ * decides whether the old task's local progress can be reused.
+ *
+ * <p>The default implementation preserves the per-metadata behavior for
handlers that do not
+ * need batch processing.
+ */
+ default boolean handlePipeMetaChanges(
+ final List<ByteBuffer> pipeMetas,
+ final List<TPushPipeMetaRespExceptionMessage> exceptionMessages)
+ throws Exception {
+ for (final ByteBuffer pipeMeta : pipeMetas) {
+ final TPushPipeMetaRespExceptionMessage message =
handleSinglePipeMeta(pipeMeta);
+ if (message != null) {
+ exceptionMessages.add(message);
+ }
+ }
+ return true;
+ }
Review Comment:
Documented in 5fb019b8d2c. The boolean is a batch-level lock result: true
means the batch was processed; false is reserved for failure to acquire the
PipeTaskAgent write lock before the metadata handling timeout. Individual pipe
failures are added to exceptionMessages rather than represented by false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
