@SuppressWarnings({
    "squid:S3776"
  }) // ignore Cognitive Complexity of methods should not be too high
  private <T> void insertByGroup(
      Map<SessionConnection, T> insertGroup, InsertConsumer<T> insertConsumer)
      throws IoTDBConnectionException, StatementExecutionException {
    List<CompletableFuture<Void>> completableFutures =
        insertGroup.entrySet().stream()
            .map(
                entry -> {
                  SessionConnection connection = entry.getKey();
                  T insertReq = entry.getValue();
                  return CompletableFuture.runAsync(
                      () -> {
                        try {
                          insertConsumer.insert(connection, insertReq);
                        } catch (RedirectException e) {
                          e.getDeviceEndPointMap().forEach(this::handleRedirection);
                        } catch (StatementExecutionException e) {
                          throw new CompletionException(e);
                        } catch (IoTDBConnectionException e) {
                          // remove the broken session
                          removeBrokenSessionConnection(connection);
                          try {
                            insertConsumer.insert(defaultSessionConnection, insertReq);
                          } catch (IoTDBConnectionException | StatementExecutionException ex) {
                            throw new CompletionException(ex);
                          } catch (RedirectException ignored) {
                          }
                        }
                      },
                      OPERATION_EXECUTOR);
                })
            .collect(Collectors.toList());

    StringBuilder errMsgBuilder = new StringBuilder();
    for (CompletableFuture<Void> completableFuture : completableFutures) {
      try {
        completableFuture.join();
      } catch (CompletionException completionException) {
        Throwable cause = completionException.getCause();
        logger.error("Meet error when async insert!", cause);
        if (cause instanceof IoTDBConnectionException) {
          throw (IoTDBConnectionException) cause;
        } else {
          if (errMsgBuilder.length() > 0) {
            errMsgBuilder.append(";");
          }
          errMsgBuilder.append(cause.getMessage());
        }
      }
    }
    if (errMsgBuilder.length() > 0) {
      throw new StatementExecutionException(errMsgBuilder.toString());
    }
  }
@param insertGroup connection to request map
@param insertConsumer insert function
@param <T>
    <ul>
      <li>{@link TSInsertRecordsReq}
      <li>{@link TSInsertStringRecordsReq}
      <li>{@link TSInsertTabletsReq}
    </ul>
iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java