public void handle(
      final TSStatus status, final String exceptionMessage, final String recordMessage) {

    if (RetryUtils.needRetryForConsensus(status.getCode())) {
      LOGGER.info("IoTConsensusV2: will retry with increasing interval. status: {}", status);
      throw new PipeConsensusRetryWithIncreasingIntervalException(
          exceptionMessage, Integer.MAX_VALUE);
    }

    if (RetryUtils.notNeedRetryForConsensus(status.getCode())) {
      LOGGER.info("IoTConsensusV2: will not retry. status: {}", status);
      return;
    }

    switch (status.getCode()) {
      case 200: // SUCCESS_STATUS
      case 400: // REDIRECTION_RECOMMEND
        {
          return;
        }

      case 1809: // PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION
        {
          LOGGER.info("Idempotent conflict exception: will be ignored. status: {}", status);
          return;
        }

      case 1808: // PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION
        {
          LOGGER.info("Temporary unavailable exception: will retry forever. status: {}", status);
          throw new PipeRuntimeConnectorCriticalException(exceptionMessage);
        }

      case 1810: // PIPE_RECEIVER_USER_CONFLICT_EXCEPTION
        if (!isRetryAllowedWhenConflictOccurs) {
          LOGGER.warn(
              "User conflict exception: will be ignored because retry is not allowed. event: {}. status: {}",
              shouldRecordIgnoredDataWhenConflictOccurs ? recordMessage : "not recorded",
              status);
          return;
        }

        synchronized (this) {
          recordExceptionStatusIfNecessary(recordMessage);

          if (exceptionEventHasBeenRetried.get()
              && System.currentTimeMillis() - exceptionFirstEncounteredTime.get()
                  > retryMaxMillisWhenConflictOccurs) {
            LOGGER.warn(
                "User conflict exception: retry timeout. will be ignored. event: {}. status: {}",
                shouldRecordIgnoredDataWhenConflictOccurs ? recordMessage : "not recorded",
                status);
            resetExceptionStatus();
            return;
          }

          LOGGER.warn(
              "User conflict exception: will retry {}. status: {}",
              retryMaxMillisWhenConflictOccurs == Long.MAX_VALUE
                  ? "forever"
                  : "for at least "
                      + (retryMaxMillisWhenConflictOccurs
                              + exceptionFirstEncounteredTime.get()
                              - System.currentTimeMillis())
                          / 1000.0
                      + " seconds",
              status);
          exceptionEventHasBeenRetried.set(true);
          throw new PipeRuntimeConnectorRetryTimesConfigurableException(
              exceptionMessage,
              (int)
                  Math.max(
                      PipeSubtask.MAX_RETRY_TIMES,
                      Math.min(CONFLICT_RETRY_MAX_TIMES, retryMaxMillisWhenConflictOccurs * 1.1)));
        }

      case 803: // NO_PERMISSION
        if (skipIfNoPrivileges) {
          return;
        }

        synchronized (this) {
          recordExceptionStatusIfNecessary(recordMessage);

          if (exceptionEventHasBeenRetried.get()
              && System.currentTimeMillis() - exceptionFirstEncounteredTime.get()
                  > retryMaxMillisWhenOtherExceptionsOccur) {
            LOGGER.warn(
                "No permission: retry timeout. will be ignored. event: {}. status: {}",
                shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
                status);
            resetExceptionStatus();
            return;
          }

          LOGGER.warn(
              "No permission: will retry {}. status: {}",
              retryMaxMillisWhenOtherExceptionsOccur == Long.MAX_VALUE
                  ? "forever"
                  : "for at least "
                      + (retryMaxMillisWhenOtherExceptionsOccur
                              + exceptionFirstEncounteredTime.get()
                              - System.currentTimeMillis())
                          / 1000.0
                      + " seconds",
              status);
          exceptionEventHasBeenRetried.set(true);
          throw new PipeRuntimeConnectorRetryTimesConfigurableException(
              exceptionMessage,
              (int)
                  Math.max(
                      PipeSubtask.MAX_RETRY_TIMES,
                      Math.min(
                          CONFLICT_RETRY_MAX_TIMES, retryMaxMillisWhenOtherExceptionsOccur * 1.1)));
        }

      default: // Other exceptions
        synchronized (this) {
          recordExceptionStatusIfNecessary(recordMessage);

          if (exceptionEventHasBeenRetried.get()
              && System.currentTimeMillis() - exceptionFirstEncounteredTime.get()
                  > retryMaxMillisWhenOtherExceptionsOccur) {
            LOGGER.warn(
                "Unclassified exception: retry timeout. will be ignored. event: {}. status: {}",
                shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
                status);
            resetExceptionStatus();
            return;
          }

          LOGGER.warn(
              "Unclassified exception: will retry {}. status: {}",
              retryMaxMillisWhenOtherExceptionsOccur == Long.MAX_VALUE
                  ? "forever"
                  : "for at least "
                      + (retryMaxMillisWhenOtherExceptionsOccur
                              + exceptionFirstEncounteredTime.get()
                              - System.currentTimeMillis())
                          / 1000.0
                      + " seconds",
              status);
          exceptionEventHasBeenRetried.set(true);
          throw new PipeRuntimeConnectorRetryTimesConfigurableException(
              exceptionMessage,
              (int)
                  Math.max(
                      PipeSubtask.MAX_RETRY_TIMES,
                      Math.min(
                          CONFLICT_RETRY_MAX_TIMES, retryMaxMillisWhenOtherExceptionsOccur * 1.1)));
        }
    }
  }
Handle {@link TSStatus} returned by receiver. Do nothing if ignore the {@link Event}, and throw
exception if retry the {@link Event}. Upper class must ensure that the method is invoked only
by a single thread.

@throws PipeException to retry the current {@link Event}
@param status the {@link TSStatus} to judge
@param exceptionMessage The exception message to throw
@param recordMessage The message to record an ignored {@link Event}, the caller should assure
    that the same {@link Event} generates always the same record message, for instance, do not
    put any time-related info here
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java