@Override
    public String getMetadataKey(MetaField metaField) {
        String metadataKey;
        switch (metaField) {
            case TABLE_NAME:
                metadataKey = "value.table";
                break;
            case DATABASE_NAME:
                metadataKey = "value.database";
                break;
            case SQL_TYPE:
                metadataKey = "value.sql-type";
                break;
            case PK_NAMES:
                metadataKey = "value.pk-names";
                break;
            case TS:
                metadataKey = "value.ingestion-timestamp";
                break;
            case OP_TS:
                metadataKey = "value.event-timestamp";
                break;
            case OP_TYPE:
                metadataKey = "value.type";
                break;
            case IS_DDL:
                metadataKey = "value.is-ddl";
                break;
            case MYSQL_TYPE:
                metadataKey = "value.mysql-type";
                break;
            case BATCH_ID:
                metadataKey = "value.batch-id";
                break;
            case UPDATE_BEFORE:
                metadataKey = "value.update-before";
                break;
            case KEY:
                metadataKey = "key";
                break;
            case VALUE:
                metadataKey = "value";
                break;
            case HEADERS:
                metadataKey = "headers";
                break;
            case HEADERS_TO_JSON_STR:
                metadataKey = "headers_to_json_str";
                break;
            case PARTITION:
                metadataKey = "partition";
                break;
            case OFFSET:
                metadataKey = "offset";
                break;
            case TIMESTAMP:
                metadataKey = "timestamp";
                break;
            case AUDIT_DATA_TIME:
                if (format instanceof InLongMsgFormat) {
                    metadataKey = INLONG_MSG_AUDIT_TIME;
                } else {
                    metadataKey = CONSUME_AUDIT_TIME;
                }
                break;
            default:
                throw new UnsupportedOperationException(String.format("Unsupport meta field for %s: %s",
                        this.getClass().getSimpleName(), metaField));
        }
        return metadataKey;
    }
true is upsert kafka connector
false is kafka connector
@return Boolean variable that decides connector option
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/KafkaExtractNode.java