You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Caideyipi (via GitHub)" <gi...@apache.org> on 2026/06/15 06:32:04 UTC
[PR] Pipe: improve progress coverage checks [iotdb]
Caideyipi opened a new pull request, #17940:
URL: https://github.com/apache/iotdb/pull/17940
## Description
### Progress coverage
Add `ProgressIndex.isEqualOrAfter(...)` and use it for pipe progress coverage checks. Historical TsFile selection now checks whether the start index already covers the resource progress by reusing `updateToMinimumEqualOrIsAfterProgressIndex`, which handles Hybrid/Recover partial-dimension cases consistently.
### Diagnostics
Add historical TsFile selection summary counters for progress-uncovered selection, unclosed/closing selection, time/path filtering, covered skips, deleted/generated-by-pipe skips, and pin failures. Shutdown progress persistence now logs pipe/meta counts, meta size, heartbeat timing, and reports clearly when shutdown progress is not confirmed by ConfigNode before the deadline.
### Tests
Add Hybrid/Recover coverage cases for historical TsFile selection and realtime TsFile epoch progress checks.
Validation:
- `.\mvnw.cmd -Ddevelocity.off=true -pl iotdb-core/node-commons,iotdb-core/datanode spotless:apply`
- `git diff --check origin/master`
Targeted tests were attempted locally, but DataNode test compilation is currently blocked by unrelated existing errors in generated query/parser and queue classes.
<hr>
This PR has:
- [x] been self-reviewed.
- [x] added Javadocs for most classes and all non-trivial methods.
- [x] added comments explaining the why and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage.
<hr>
##### Key changed/added classes (or packages if there are too many classes) in this PR
- `ProgressIndex`
- `PipeHistoricalDataRegionTsFileAndDeletionSource`
- `PipeTsFileEpochProgressIndexKeeper`
- `PipeDataNodeTaskAgent`
- `DataNodeShutdownHook`
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve progress coverage checks [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #17940:
URL: https://github.com/apache/iotdb/pull/17940#discussion_r3411959698
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -623,25 +626,140 @@ public void runPipeTasks(
///////////////////////// Shutdown Logic /////////////////////////
+ public long getShutdownProgressPersistTimeoutInMs() {
+ return Math.max(
+ 1_000L,
+ (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS()
+ + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS());
+ }
+
+ public boolean persistAllProgressIndex(final long timeoutInMs) {
+ final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs);
+ final long startTime = System.currentTimeMillis();
+ final AtomicBoolean isConfirmed = new AtomicBoolean(false);
+ final Thread persistThread =
+ new Thread(
+ () -> isConfirmed.set(persistAllProgressIndexInternal()),
+ ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist");
+ persistThread.setDaemon(true);
+
+ LOGGER.info(
+ "Start to persist all pipe progress indexes during shutdown, pipe count {}, deadline {} ms",
+ getPipeCount(),
+ normalizedTimeoutInMs);
Review Comment:
Done. All newly added shutdown progress logs now use DataNodePipeMessages, with English and Chinese i18n entries.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -130,6 +131,8 @@
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskAgent.class);
+ private static final String SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE =
+ "\u672c\u6b21 shutdown progress \u672a\u786e\u8ba4\u843d\u5230 ConfigNode.";
Review Comment:
Done. Removed the hard-coded Chinese suffix and moved the shutdown progress confirmation message into DataNodePipeMessages for both en/zh locales.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -623,25 +626,140 @@ public void runPipeTasks(
///////////////////////// Shutdown Logic /////////////////////////
+ public long getShutdownProgressPersistTimeoutInMs() {
+ return Math.max(
+ 1_000L,
+ (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS()
+ + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS());
+ }
+
+ public boolean persistAllProgressIndex(final long timeoutInMs) {
+ final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs);
+ final long startTime = System.currentTimeMillis();
+ final AtomicBoolean isConfirmed = new AtomicBoolean(false);
+ final Thread persistThread =
+ new Thread(
+ () -> isConfirmed.set(persistAllProgressIndexInternal()),
+ ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist");
+ persistThread.setDaemon(true);
+
+ LOGGER.info(
+ "Start to persist all pipe progress indexes during shutdown, pipe count {}, deadline {} ms",
+ getPipeCount(),
+ normalizedTimeoutInMs);
+ persistThread.start();
+ try {
+ final long deadlineInMs = startTime + normalizedTimeoutInMs;
+ while (persistThread.isAlive()) {
+ final long remainingTimeInMs = deadlineInMs - System.currentTimeMillis();
+ if (remainingTimeInMs <= 0) {
+ break;
+ }
+ persistThread.join(remainingTimeInMs);
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Interrupted while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ e);
+ return false;
+ }
+
+ if (persistThread.isAlive()) {
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Timed out after {} ms while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ return false;
+ }
+
+ if (!isConfirmed.get()) {
+ LOGGER.warn(
+ "Failed to persist all pipe progress indexes during shutdown within {} ms. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ }
+ return isConfirmed.get();
+ }
+
public void persistAllProgressIndex() {
- try (final ConfigNodeClient configNodeClient =
- ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- // Send request to some API server
+ persistAllProgressIndex(getShutdownProgressPersistTimeoutInMs());
+ }
+
+ private boolean persistAllProgressIndexInternal() {
+ final long collectStartTime = System.currentTimeMillis();
+ final int pipeCount = getPipeCount();
+ try {
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>());
collectPipeMetaList(new TPipeHeartbeatReq(Long.MIN_VALUE), resp);
+ final int pipeMetaCount = resp.getPipeMetaList().size();
+ final int pipeMetaSizeInBytes =
+ resp.getPipeMetaList().stream()
+ .filter(Objects::nonNull)
+ .mapToInt(ByteBuffer::remaining)
+ .sum();
+ LOGGER.info(
+ "Collected pipe metas for shutdown progress persist, pipe count {}, pipe meta count {}, "
+ + "pipe meta size {} bytes, took {} ms",
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes,
+ System.currentTimeMillis() - collectStartTime);
+
if (resp.getPipeMetaList().isEmpty()) {
- return;
+ if (pipeCount != 0) {
+ LOGGER.warn(
+ "Collected empty pipe metas for {} pipes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ pipeCount);
+ return false;
+ }
+ return true;
}
- final TSStatus result =
- configNodeClient.pushHeartbeat(
- IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp);
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_PERSIST_PROGRESS_INDEX_TO_CONFIGNODE, result);
- } else {
- LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_PERSISTED_ALL_PIPE_S_INFO_TO);
+
+ try (final ConfigNodeClient configNodeClient =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ LOGGER.info(
+ "Start to pushHeartbeat shutdown pipe meta to ConfigNode, dataNode id {}, pipe count {}, "
+ + "pipe meta count {}, pipe meta size {} bytes",
+ IoTDBDescriptor.getInstance().getConfig().getDataNodeId(),
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes);
+ final long pushStartTime = System.currentTimeMillis();
+ final TSStatus result =
+ configNodeClient.pushHeartbeat(
+ IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp);
+ final long pushCostTime = System.currentTimeMillis() - pushStartTime;
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) {
+ LOGGER.warn(DataNodePipeMessages.FAILED_TO_PERSIST_PROGRESS_INDEX_TO_CONFIGNODE, result);
+ LOGGER.warn(
+ "Failed to pushHeartbeat shutdown pipe meta to ConfigNode, status {}, took {} ms. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ result,
+ pushCostTime);
+ return false;
+ } else {
+ LOGGER.info(
+ "Successfully finished pushHeartbeat shutdown pipe meta to ConfigNode, pipe count {}, "
+ + "pipe meta count {}, pipe meta size {} bytes, took {} ms",
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes,
+ pushCostTime);
+ LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_PERSISTED_ALL_PIPE_S_INFO_TO);
+ return true;
+ }
}
} catch (final Exception e) {
- LOGGER.warn(e.getMessage());
+ LOGGER.warn(
+ "Exception occurred while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ e);
+ return false;
Review Comment:
Done. The join interruption now restores the interrupt flag and logs at info without interrupting the daemon persist thread.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -623,25 +626,140 @@ public void runPipeTasks(
///////////////////////// Shutdown Logic /////////////////////////
+ public long getShutdownProgressPersistTimeoutInMs() {
+ return Math.max(
+ 1_000L,
+ (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS()
+ + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS());
+ }
+
+ public boolean persistAllProgressIndex(final long timeoutInMs) {
+ final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs);
+ final long startTime = System.currentTimeMillis();
+ final AtomicBoolean isConfirmed = new AtomicBoolean(false);
+ final Thread persistThread =
+ new Thread(
+ () -> isConfirmed.set(persistAllProgressIndexInternal()),
+ ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist");
+ persistThread.setDaemon(true);
+
+ LOGGER.info(
+ "Start to persist all pipe progress indexes during shutdown, pipe count {}, deadline {} ms",
+ getPipeCount(),
+ normalizedTimeoutInMs);
+ persistThread.start();
+ try {
+ final long deadlineInMs = startTime + normalizedTimeoutInMs;
+ while (persistThread.isAlive()) {
+ final long remainingTimeInMs = deadlineInMs - System.currentTimeMillis();
+ if (remainingTimeInMs <= 0) {
+ break;
+ }
+ persistThread.join(remainingTimeInMs);
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Interrupted while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ e);
+ return false;
+ }
+
+ if (persistThread.isAlive()) {
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Timed out after {} ms while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ return false;
+ }
+
+ if (!isConfirmed.get()) {
+ LOGGER.warn(
+ "Failed to persist all pipe progress indexes during shutdown within {} ms. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ }
+ return isConfirmed.get();
+ }
+
public void persistAllProgressIndex() {
- try (final ConfigNodeClient configNodeClient =
- ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- // Send request to some API server
+ persistAllProgressIndex(getShutdownProgressPersistTimeoutInMs());
+ }
+
+ private boolean persistAllProgressIndexInternal() {
+ final long collectStartTime = System.currentTimeMillis();
+ final int pipeCount = getPipeCount();
+ try {
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>());
collectPipeMetaList(new TPipeHeartbeatReq(Long.MIN_VALUE), resp);
+ final int pipeMetaCount = resp.getPipeMetaList().size();
+ final int pipeMetaSizeInBytes =
+ resp.getPipeMetaList().stream()
+ .filter(Objects::nonNull)
+ .mapToInt(ByteBuffer::remaining)
+ .sum();
+ LOGGER.info(
+ "Collected pipe metas for shutdown progress persist, pipe count {}, pipe meta count {}, "
+ + "pipe meta size {} bytes, took {} ms",
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes,
+ System.currentTimeMillis() - collectStartTime);
+
if (resp.getPipeMetaList().isEmpty()) {
- return;
+ if (pipeCount != 0) {
+ LOGGER.warn(
+ "Collected empty pipe metas for {} pipes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ pipeCount);
Review Comment:
Done. Changed this log to info. The method still returns false when pipeCount is non-zero but no meta is collected, so the shutdown hook keeps the single final warning that progress was not confirmed.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve progress coverage checks [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #17940:
URL: https://github.com/apache/iotdb/pull/17940
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve progress coverage checks [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #17940:
URL: https://github.com/apache/iotdb/pull/17940#discussion_r3411495562
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -130,6 +131,8 @@
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskAgent.class);
+ private static final String SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE =
+ "\u672c\u6b21 shutdown progress \u672a\u786e\u8ba4\u843d\u5230 ConfigNode.";
Review Comment:
What is this?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -623,25 +626,140 @@ public void runPipeTasks(
///////////////////////// Shutdown Logic /////////////////////////
+ public long getShutdownProgressPersistTimeoutInMs() {
+ return Math.max(
+ 1_000L,
+ (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS()
+ + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS());
+ }
+
+ public boolean persistAllProgressIndex(final long timeoutInMs) {
+ final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs);
+ final long startTime = System.currentTimeMillis();
+ final AtomicBoolean isConfirmed = new AtomicBoolean(false);
+ final Thread persistThread =
+ new Thread(
+ () -> isConfirmed.set(persistAllProgressIndexInternal()),
+ ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist");
+ persistThread.setDaemon(true);
+
+ LOGGER.info(
+ "Start to persist all pipe progress indexes during shutdown, pipe count {}, deadline {} ms",
+ getPipeCount(),
+ normalizedTimeoutInMs);
+ persistThread.start();
+ try {
+ final long deadlineInMs = startTime + normalizedTimeoutInMs;
+ while (persistThread.isAlive()) {
+ final long remainingTimeInMs = deadlineInMs - System.currentTimeMillis();
+ if (remainingTimeInMs <= 0) {
+ break;
+ }
+ persistThread.join(remainingTimeInMs);
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Interrupted while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ e);
+ return false;
+ }
+
+ if (persistThread.isAlive()) {
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Timed out after {} ms while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ return false;
+ }
+
+ if (!isConfirmed.get()) {
+ LOGGER.warn(
+ "Failed to persist all pipe progress indexes during shutdown within {} ms. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ }
+ return isConfirmed.get();
+ }
+
public void persistAllProgressIndex() {
- try (final ConfigNodeClient configNodeClient =
- ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- // Send request to some API server
+ persistAllProgressIndex(getShutdownProgressPersistTimeoutInMs());
+ }
+
+ private boolean persistAllProgressIndexInternal() {
+ final long collectStartTime = System.currentTimeMillis();
+ final int pipeCount = getPipeCount();
+ try {
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>());
collectPipeMetaList(new TPipeHeartbeatReq(Long.MIN_VALUE), resp);
+ final int pipeMetaCount = resp.getPipeMetaList().size();
+ final int pipeMetaSizeInBytes =
+ resp.getPipeMetaList().stream()
+ .filter(Objects::nonNull)
+ .mapToInt(ByteBuffer::remaining)
+ .sum();
+ LOGGER.info(
+ "Collected pipe metas for shutdown progress persist, pipe count {}, pipe meta count {}, "
+ + "pipe meta size {} bytes, took {} ms",
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes,
+ System.currentTimeMillis() - collectStartTime);
+
if (resp.getPipeMetaList().isEmpty()) {
- return;
+ if (pipeCount != 0) {
+ LOGGER.warn(
+ "Collected empty pipe metas for {} pipes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ pipeCount);
+ return false;
+ }
+ return true;
}
- final TSStatus result =
- configNodeClient.pushHeartbeat(
- IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp);
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_PERSIST_PROGRESS_INDEX_TO_CONFIGNODE, result);
- } else {
- LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_PERSISTED_ALL_PIPE_S_INFO_TO);
+
+ try (final ConfigNodeClient configNodeClient =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ LOGGER.info(
+ "Start to pushHeartbeat shutdown pipe meta to ConfigNode, dataNode id {}, pipe count {}, "
+ + "pipe meta count {}, pipe meta size {} bytes",
+ IoTDBDescriptor.getInstance().getConfig().getDataNodeId(),
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes);
+ final long pushStartTime = System.currentTimeMillis();
+ final TSStatus result =
+ configNodeClient.pushHeartbeat(
+ IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp);
+ final long pushCostTime = System.currentTimeMillis() - pushStartTime;
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) {
+ LOGGER.warn(DataNodePipeMessages.FAILED_TO_PERSIST_PROGRESS_INDEX_TO_CONFIGNODE, result);
+ LOGGER.warn(
+ "Failed to pushHeartbeat shutdown pipe meta to ConfigNode, status {}, took {} ms. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ result,
+ pushCostTime);
+ return false;
+ } else {
+ LOGGER.info(
+ "Successfully finished pushHeartbeat shutdown pipe meta to ConfigNode, pipe count {}, "
+ + "pipe meta count {}, pipe meta size {} bytes, took {} ms",
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes,
+ pushCostTime);
+ LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_PERSISTED_ALL_PIPE_S_INFO_TO);
+ return true;
+ }
}
} catch (final Exception e) {
- LOGGER.warn(e.getMessage());
+ LOGGER.warn(
+ "Exception occurred while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ e);
+ return false;
Review Comment:
May ignore interruption here
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -623,25 +626,140 @@ public void runPipeTasks(
///////////////////////// Shutdown Logic /////////////////////////
+ public long getShutdownProgressPersistTimeoutInMs() {
+ return Math.max(
+ 1_000L,
+ (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS()
+ + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS());
+ }
+
+ public boolean persistAllProgressIndex(final long timeoutInMs) {
+ final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs);
+ final long startTime = System.currentTimeMillis();
+ final AtomicBoolean isConfirmed = new AtomicBoolean(false);
+ final Thread persistThread =
+ new Thread(
+ () -> isConfirmed.set(persistAllProgressIndexInternal()),
+ ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist");
+ persistThread.setDaemon(true);
+
+ LOGGER.info(
+ "Start to persist all pipe progress indexes during shutdown, pipe count {}, deadline {} ms",
+ getPipeCount(),
+ normalizedTimeoutInMs);
+ persistThread.start();
+ try {
+ final long deadlineInMs = startTime + normalizedTimeoutInMs;
+ while (persistThread.isAlive()) {
+ final long remainingTimeInMs = deadlineInMs - System.currentTimeMillis();
+ if (remainingTimeInMs <= 0) {
+ break;
+ }
+ persistThread.join(remainingTimeInMs);
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Interrupted while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ e);
+ return false;
+ }
+
+ if (persistThread.isAlive()) {
+ persistThread.interrupt();
+ LOGGER.warn(
+ "Timed out after {} ms while persisting all pipe progress indexes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ return false;
+ }
+
+ if (!isConfirmed.get()) {
+ LOGGER.warn(
+ "Failed to persist all pipe progress indexes during shutdown within {} ms. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ System.currentTimeMillis() - startTime);
+ }
+ return isConfirmed.get();
+ }
+
public void persistAllProgressIndex() {
- try (final ConfigNodeClient configNodeClient =
- ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- // Send request to some API server
+ persistAllProgressIndex(getShutdownProgressPersistTimeoutInMs());
+ }
+
+ private boolean persistAllProgressIndexInternal() {
+ final long collectStartTime = System.currentTimeMillis();
+ final int pipeCount = getPipeCount();
+ try {
final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>());
collectPipeMetaList(new TPipeHeartbeatReq(Long.MIN_VALUE), resp);
+ final int pipeMetaCount = resp.getPipeMetaList().size();
+ final int pipeMetaSizeInBytes =
+ resp.getPipeMetaList().stream()
+ .filter(Objects::nonNull)
+ .mapToInt(ByteBuffer::remaining)
+ .sum();
+ LOGGER.info(
+ "Collected pipe metas for shutdown progress persist, pipe count {}, pipe meta count {}, "
+ + "pipe meta size {} bytes, took {} ms",
+ pipeCount,
+ pipeMetaCount,
+ pipeMetaSizeInBytes,
+ System.currentTimeMillis() - collectStartTime);
+
if (resp.getPipeMetaList().isEmpty()) {
- return;
+ if (pipeCount != 0) {
+ LOGGER.warn(
+ "Collected empty pipe metas for {} pipes during shutdown. "
+ + SHUTDOWN_PROGRESS_NOT_CONFIRMED_MESSAGE,
+ pipeCount);
Review Comment:
Necessary to warn?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -623,25 +626,140 @@ public void runPipeTasks(
///////////////////////// Shutdown Logic /////////////////////////
+ public long getShutdownProgressPersistTimeoutInMs() {
+ return Math.max(
+ 1_000L,
+ (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS()
+ + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS());
+ }
+
+ public boolean persistAllProgressIndex(final long timeoutInMs) {
+ final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs);
+ final long startTime = System.currentTimeMillis();
+ final AtomicBoolean isConfirmed = new AtomicBoolean(false);
+ final Thread persistThread =
+ new Thread(
+ () -> isConfirmed.set(persistAllProgressIndexInternal()),
+ ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist");
+ persistThread.setDaemon(true);
+
+ LOGGER.info(
+ "Start to persist all pipe progress indexes during shutdown, pipe count {}, deadline {} ms",
+ getPipeCount(),
+ normalizedTimeoutInMs);
Review Comment:
i18n, the same below.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org