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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
