|
99 | 99 | import java.util.concurrent.Future; |
100 | 100 | import java.util.concurrent.ThreadPoolExecutor; |
101 | 101 | import java.util.concurrent.TimeUnit; |
| 102 | +import java.util.concurrent.atomic.AtomicBoolean; |
102 | 103 | import java.util.function.Consumer; |
103 | 104 | import java.util.stream.Collectors; |
104 | 105 |
|
@@ -623,25 +624,129 @@ public void runPipeTasks( |
623 | 624 |
|
624 | 625 | ///////////////////////// Shutdown Logic ///////////////////////// |
625 | 626 |
|
| 627 | + public long getShutdownProgressPersistTimeoutInMs() { |
| 628 | + return Math.max( |
| 629 | + 1_000L, |
| 630 | + (long) CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS() |
| 631 | + + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS()); |
| 632 | + } |
| 633 | + |
| 634 | + public boolean persistAllProgressIndex(final long timeoutInMs) { |
| 635 | + final long normalizedTimeoutInMs = Math.max(1L, timeoutInMs); |
| 636 | + final long startTime = System.currentTimeMillis(); |
| 637 | + final AtomicBoolean isConfirmed = new AtomicBoolean(false); |
| 638 | + final Thread persistThread = |
| 639 | + new Thread( |
| 640 | + () -> isConfirmed.set(persistAllProgressIndexInternal()), |
| 641 | + ThreadName.PIPE_RUNTIME_META_SYNCER.getName() + "-Shutdown-Persist"); |
| 642 | + persistThread.setDaemon(true); |
| 643 | + |
| 644 | + LOGGER.info( |
| 645 | + DataNodePipeMessages.START_TO_PERSIST_ALL_PIPE_PROGRESS_INDEXES_DURING_SHUTDOWN, |
| 646 | + getPipeCount(), |
| 647 | + normalizedTimeoutInMs); |
| 648 | + persistThread.start(); |
| 649 | + try { |
| 650 | + final long deadlineInMs = startTime + normalizedTimeoutInMs; |
| 651 | + while (persistThread.isAlive()) { |
| 652 | + final long remainingTimeInMs = deadlineInMs - System.currentTimeMillis(); |
| 653 | + if (remainingTimeInMs <= 0) { |
| 654 | + break; |
| 655 | + } |
| 656 | + persistThread.join(remainingTimeInMs); |
| 657 | + } |
| 658 | + } catch (final InterruptedException e) { |
| 659 | + Thread.currentThread().interrupt(); |
| 660 | + LOGGER.info( |
| 661 | + DataNodePipeMessages |
| 662 | + .INTERRUPTED_WHILE_PERSISTING_ALL_PIPE_PROGRESS_INDEXES_DURING_SHUTDOWN); |
| 663 | + return false; |
| 664 | + } |
| 665 | + |
| 666 | + if (persistThread.isAlive()) { |
| 667 | + LOGGER.warn( |
| 668 | + DataNodePipeMessages.TIMED_OUT_WHILE_PERSISTING_ALL_PIPE_PROGRESS_INDEXES_DURING_SHUTDOWN, |
| 669 | + System.currentTimeMillis() - startTime); |
| 670 | + return false; |
| 671 | + } |
| 672 | + |
| 673 | + if (!isConfirmed.get()) { |
| 674 | + LOGGER.warn( |
| 675 | + DataNodePipeMessages.FAILED_TO_PERSIST_ALL_PIPE_PROGRESS_INDEXES_DURING_SHUTDOWN, |
| 676 | + System.currentTimeMillis() - startTime); |
| 677 | + } |
| 678 | + return isConfirmed.get(); |
| 679 | + } |
| 680 | + |
626 | 681 | public void persistAllProgressIndex() { |
627 | | - try (final ConfigNodeClient configNodeClient = |
628 | | - ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { |
629 | | - // Send request to some API server |
| 682 | + persistAllProgressIndex(getShutdownProgressPersistTimeoutInMs()); |
| 683 | + } |
| 684 | + |
| 685 | + private boolean persistAllProgressIndexInternal() { |
| 686 | + final long collectStartTime = System.currentTimeMillis(); |
| 687 | + final int pipeCount = getPipeCount(); |
| 688 | + try { |
630 | 689 | final TPipeHeartbeatResp resp = new TPipeHeartbeatResp(new ArrayList<>()); |
631 | 690 | collectPipeMetaList(new TPipeHeartbeatReq(Long.MIN_VALUE), resp); |
| 691 | + final int pipeMetaCount = resp.getPipeMetaList().size(); |
| 692 | + final int pipeMetaSizeInBytes = |
| 693 | + resp.getPipeMetaList().stream() |
| 694 | + .filter(Objects::nonNull) |
| 695 | + .mapToInt(ByteBuffer::remaining) |
| 696 | + .sum(); |
| 697 | + LOGGER.info( |
| 698 | + DataNodePipeMessages.COLLECTED_PIPE_METAS_FOR_SHUTDOWN_PROGRESS_PERSIST, |
| 699 | + pipeCount, |
| 700 | + pipeMetaCount, |
| 701 | + pipeMetaSizeInBytes, |
| 702 | + System.currentTimeMillis() - collectStartTime); |
| 703 | + |
632 | 704 | if (resp.getPipeMetaList().isEmpty()) { |
633 | | - return; |
| 705 | + if (pipeCount != 0) { |
| 706 | + LOGGER.info(DataNodePipeMessages.COLLECTED_EMPTY_PIPE_METAS_DURING_SHUTDOWN, pipeCount); |
| 707 | + return false; |
| 708 | + } |
| 709 | + return true; |
634 | 710 | } |
635 | | - final TSStatus result = |
636 | | - configNodeClient.pushHeartbeat( |
637 | | - IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp); |
638 | | - if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) { |
639 | | - LOGGER.warn(DataNodePipeMessages.FAILED_TO_PERSIST_PROGRESS_INDEX_TO_CONFIGNODE, result); |
640 | | - } else { |
641 | | - LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_PERSISTED_ALL_PIPE_S_INFO_TO); |
| 711 | + |
| 712 | + try (final ConfigNodeClient configNodeClient = |
| 713 | + ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { |
| 714 | + LOGGER.info( |
| 715 | + DataNodePipeMessages.START_TO_PUSH_HEARTBEAT_SHUTDOWN_PIPE_META_TO_CONFIGNODE, |
| 716 | + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), |
| 717 | + pipeCount, |
| 718 | + pipeMetaCount, |
| 719 | + pipeMetaSizeInBytes); |
| 720 | + final long pushStartTime = System.currentTimeMillis(); |
| 721 | + final TSStatus result = |
| 722 | + configNodeClient.pushHeartbeat( |
| 723 | + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp); |
| 724 | + final long pushCostTime = System.currentTimeMillis() - pushStartTime; |
| 725 | + if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) { |
| 726 | + LOGGER.warn(DataNodePipeMessages.FAILED_TO_PERSIST_PROGRESS_INDEX_TO_CONFIGNODE, result); |
| 727 | + LOGGER.warn( |
| 728 | + DataNodePipeMessages.FAILED_TO_PUSH_HEARTBEAT_SHUTDOWN_PIPE_META_TO_CONFIGNODE, |
| 729 | + result, |
| 730 | + pushCostTime); |
| 731 | + return false; |
| 732 | + } else { |
| 733 | + LOGGER.info( |
| 734 | + DataNodePipeMessages |
| 735 | + .SUCCESSFULLY_FINISHED_PUSH_HEARTBEAT_SHUTDOWN_PIPE_META_TO_CONFIGNODE, |
| 736 | + pipeCount, |
| 737 | + pipeMetaCount, |
| 738 | + pipeMetaSizeInBytes, |
| 739 | + pushCostTime); |
| 740 | + LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_PERSISTED_ALL_PIPE_S_INFO_TO); |
| 741 | + return true; |
| 742 | + } |
642 | 743 | } |
643 | 744 | } catch (final Exception e) { |
644 | | - LOGGER.warn(e.getMessage()); |
| 745 | + LOGGER.warn( |
| 746 | + DataNodePipeMessages |
| 747 | + .EXCEPTION_OCCURRED_WHILE_PERSISTING_ALL_PIPE_PROGRESS_INDEXES_DURING_SHUTDOWN, |
| 748 | + e); |
| 749 | + return false; |
645 | 750 | } |
646 | 751 | } |
647 | 752 |
|
|
0 commit comments