private void establishConnection() throws IOException {
        KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(kuduMasters);
        if (null != workerCount) {
            kuduClientBuilder.workerCount(workerCount);
        }

        if (null != defaultOperationTimeoutMs) {
            kuduClientBuilder.defaultOperationTimeoutMs(defaultOperationTimeoutMs);
        }

        if (enableKrb) {
            UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(
                    principal,
                    keytab,
                    krb5conf
            );
            client = ugi.doAs(
                    (PrivilegedAction<KuduClient>) kuduClientBuilder::build);
        } else {
            client = kuduClientBuilder.build();
        }

        if (client.tableExists(tableName)) {
            table = client.openTable(tableName);
        }
        if (Objects.isNull(table)) {
            throw new IllegalArgumentException(
                    String.format("Table [%s] Open Failed , please check table exists", tableName));
        }
        LOG.info("connect kudu is succeed!");

        session = buildSessionWithFlushMode(flushMode, client);
    }
init the scheduler task of {@link KuduOutputFormat#flush()}
kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java