@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        options.put(TOPIC, topic);
        options.put(PROPERTIES_BOOTSTRAP_SERVERS, bootstrapServers);
        if (getSinkParallelism() != null) {
            options.put(SINK_PARALLELISM, getSinkParallelism().toString());
        }
        if (format instanceof JsonFormat || format instanceof AvroFormat
                || format instanceof CsvFormat || format instanceof RawFormat) {
            if (StringUtils.isEmpty(this.primaryKey)) {
                options.put(CONNECTOR, KAFKA);
                options.put(SINK_IGNORE_CHANGELOG, "true");
                options.putAll(format.generateOptions(false));
            } else {
                options.put(CONNECTOR, UPSERT_KAFKA);
                options.putAll(format.generateOptions(true));
            }
            if (format instanceof RawFormat) {
                if (sinkMultipleFormat != null) {
                    options.put(SINK_MULTIPLE_FORMAT, sinkMultipleFormat.identifier());
                }
                if (StringUtils.isNotBlank(topicPattern)) {
                    options.put(TOPIC_PATTERN, topicPattern);
                }
                if (StringUtils.isNotBlank(sinkPartitioner)) {
                    options.put(SINK_PARTITIONER, sinkPartitioner);
                }
                if (StringUtils.isNotBlank(partitionPattern)) {
                    options.put(SINK_MULTIPLE_PARTITION_PATTERN, partitionPattern);
                }
            }
        } else if (format instanceof CanalJsonFormat || format instanceof DebeziumJsonFormat) {
            options.put(CONNECTOR, KAFKA);
            options.putAll(format.generateOptions(false));
        } else {
            throw new IllegalArgumentException("kafka load Node format is IllegalArgument");
        }
        return options;
    }
Generate options for kafka connector

@return options
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/KafkaLoadNode.java