@Override
    public void invoke(DirtyData<T> dirtyData) throws Exception {
        String value;
        Map<String, String> labelMap = LabelUtils.parseLabels(dirtyData.getLabels());
        T data = dirtyData.getData();
        if (data instanceof RowData) {
            value = format((RowData) data, dirtyData.getRowType(), labelMap);
        } else if (data instanceof JsonNode) {
            value = format((JsonNode) data, labelMap);
        } else {
            // Only support csv format when the row is not a 'RowData' and 'JsonNode'
            value = FormatUtils.csvFormat(data, labelMap, fieldDelimiter);
        }
        LOGGER.debug("[{}] {}", dirtyData.getLogTag(), value);
    }
Log dirty sink that is used to print log

@param <T>
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/sink/log/LogDirtySink.java