@Override
                    public Object read(ConsumerRecord<?, ?> record) {
                        Map<String, String> headerMap = new HashMap<>();
                        for (Header header : record.headers()) {
                            headerMap.put(header.key(),
                                    new String(header.value(), StandardCharsets.UTF_8));
                        }
                        try {
                            return StringData.fromString(MAPPER.writeValueAsString(headerMap));
                        } catch (JsonProcessingException e) {
                            LOG.warn("Failed to parse headers to json string", e);
                            return null;
                        }
                    }
Flag to determine source mode. In upsert mode, it will keep the tombstone message. *
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicSource.java