Re: [PR] fix(pipe): reuse safe local progress on alter [iotdb]
jt2594838 merged PR #18423: URL: https://github.com/apache/iotdb/pull/18423 -- 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]
Re: [PR] fix(pipe): reuse safe local progress on alter [iotdb]
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
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.
+ *
+ * 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.
+ *
+ * The default implementation preserves the per-metadata behavior for
handlers that do not
+ * need batch processing.
+ */
+default boolean handlePipeMetaChanges(
+final List pipeMetas,
+final List 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]
Re: [PR] fix(pipe): reuse safe local progress on alter [iotdb]
Caideyipi commented on PR #18423: URL: https://github.com/apache/iotdb/pull/18423#issuecomment-5249692749 Added testAlterPipeDoesNotResendCommittedData in 5fb019b8d2c. It transfers two rows, deletes them on the receiver, alters the sink configuration, asserts that the deleted rows are not sent again, then writes a third row and verifies that replication remains active. The sender heartbeat interval is set to 600 seconds so the test covers stale ConfigNode metadata instead of relying on a periodic heartbeat. The targeted IT passes locally. -- 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]
Re: [PR] fix(pipe): reuse safe local progress on alter [iotdb]
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.
+ *
+ * The default implementation preserves the per-metadata behavior for
handlers that do not
+ * need batch processing.
+ */
+default boolean handlePipeMetaChanges(
+final List pipeMetas,
+final List 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
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.
+ *
+ * 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]
