public void createTable(String tableName, KuduTableInfo tableInfo) throws KuduException {
        // Parse column from tableInfo.
        List<KuduColumnInfo> columns = tableInfo.getColumns();
        List<ColumnSchema> kuduColumns = columns.stream()
                .sorted(Comparator.comparing(KuduColumnInfo::getPartitionStrategy))
                .map(this::buildColumnSchema)
                .collect(Collectors.toList());

        // Build schema.
        Schema schema = new Schema(kuduColumns);
        // Create table options: like partitions
        CreateTableOptions options = new CreateTableOptions();
        List<String> parCols = columns.stream()
                .filter(column -> PARTITION_STRATEGY_HASH.equalsIgnoreCase(column.getPartitionStrategy()))
                .map(KuduColumnInfo::getFieldName).collect(Collectors.toList());
        if (!parCols.isEmpty()) {
            Integer partitionNum = tableInfo.getBuckets();
            int buckets = partitionNum == null ? DEFAULT_BUCKETS : partitionNum;
            options.addHashPartitions(parCols, buckets);
        }

        // Create table by KuduClient.
        client.createTable(tableName, schema, options);
    }
Create table with given tableName and tableInfo.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/kudu/KuduResourceClient.java