private DataStreamSink<?> consume(
            DataStream<RowData> dataStream, boolean isBounded, DataStructureConverter converter) {
        checkAcidTable(catalogTable, identifier.toObjectPath());
        // Get dbName from with properties first.
        String dbName = jobConf.get(HiveValidator.CONNECTOR_HIVE_DATABASE);
        if (dbName == null) {
            dbName = identifier.getDatabaseName();
        }
        try (HiveMetastoreClientWrapper client =
                HiveMetastoreClientFactory.create(HiveConfUtils.create(jobConf), hiveVersion)) {

            StorageDescriptor sd;
            Properties tableProps = new Properties();
            Class hiveOutputFormatClz;
            boolean isCompressed =
                    jobConf.getBoolean(HiveConf.ConfVars.COMPRESSRESULT.varname, false);
            if (sinkMultipleEnable) {
                sd = new StorageDescriptor();
                SerDeInfo serDeInfo = new SerDeInfo();
                serDeInfo.setSerializationLib(this.serializationLib);
                sd.setSerdeInfo(serDeInfo);
                String defaultFs = jobConf.get("fs.defaultFS", "");
                sd.setLocation(defaultFs + "/tmp");
                hiveOutputFormatClz = hiveShim.getHiveOutputFormatClass(HiveIgnoreKeyTextOutputFormat.class);
            } else {
                Table table = client.getTable(dbName, identifier.getObjectName());
                sd = table.getSd();
                tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, table);
                hiveOutputFormatClz = hiveShim.getHiveOutputFormatClass(Class.forName(sd.getOutputFormat()));
            }
            HiveWriterFactory writerFactory = new HiveWriterFactory(
                    jobConf,
                    hiveOutputFormatClz,
                    sd,
                    tableSchema,
                    getPartitionKeyArray(),
                    tableProps,
                    hiveShim,
                    isCompressed,
                    sinkMultipleEnable);

            String extension =
                    Utilities.getFileExtension(
                            jobConf,
                            isCompressed,
                            (HiveOutputFormat<?, ?>) hiveOutputFormatClz.newInstance());

            OutputFileConfig.OutputFileConfigBuilder fileNamingBuilder =
                    OutputFileConfig.builder()
                            .withPartPrefix("part-" + UUID.randomUUID().toString())
                            .withPartSuffix(extension == null ? "" : extension);

            final int parallelism =
                    Optional.ofNullable(configuredParallelism).orElse(dataStream.getParallelism());
            if (isBounded) {
                OutputFileConfig fileNaming = fileNamingBuilder.build();
                return createBatchSink(dataStream, converter, sd, writerFactory, fileNaming, parallelism);
            } else {
                if (overwrite) {
                    throw new IllegalStateException("Streaming mode not support overwrite.");
                }

                return createStreamSink(
                        dataStream, sd, tableProps, writerFactory, fileNamingBuilder, parallelism);
            }
        } catch (TException e) {
            throw new CatalogException("Failed to query Hive metaStore", e);
        } catch (IOException e) {
            throw new FlinkRuntimeException("Failed to create staging dir", e);
        } catch (ClassNotFoundException e) {
            throw new FlinkHiveException("Failed to get output format class", e);
        } catch (IllegalAccessException | InstantiationException e) {
            throw new FlinkHiveException("Failed to instantiate output format instance", e);
        }
    }
Table sink to write to Hive tables.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/HiveTableSink.java