private KuduSession buildSessionWithFlushMode(String flushMode, KuduClient kuduClient) {
        KuduSession kuduSession = kuduClient.newSession();
        if (flushMode.equalsIgnoreCase(KuduTableInfo.KuduFlushMode.MANUAL_FLUSH.name())) {
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
            kuduSession.setMutationBufferSpace(mutationBufferMaxOps);
        }

        if (flushMode.equalsIgnoreCase(KuduTableInfo.KuduFlushMode.AUTO_FLUSH_SYNC.name())) {
            LOG.warn("Parameter [batchSize] will not take effect at AUTO_FLUSH_SYNC mode.");
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
        }

        if (flushMode.equalsIgnoreCase(KuduTableInfo.KuduFlushMode.AUTO_FLUSH_BACKGROUND.name())) {
            LOG.warn("Unable to determine the order of data at AUTO_FLUSH_BACKGROUND mode. Only [batchWaitInterval] will effect.");
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
            kuduSession.setFlushInterval(batchWaitInterval);
        }

        return kuduSession;
    }
According to the different flush mode, build different session. Detail see {@link SessionConfiguration.FlushMode}

@param flushMode  flush mode
@param kuduClient kudu client
@return KuduSession with flush mode
kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java