jt2594838 commented on code in PR #18423:
URL: https://github.com/apache/iotdb/pull/18423#discussion_r3746242965
##########
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:
I cannot associate `returning false` with `timeout` from the implementation
above.
Please give some explanation.
##########
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:
What is the meaning of the return value?
##########
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:
Is it possible to report the progress to the leader before the changes, so
that even the new leader may have the latest progress?
--
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]
