protected FlinkTubeMQConsumer<RowData> createTubeMQConsumer(
            DeserializationSchema<RowData> deserialization,
            TypeInformation<RowData> producedTypeInfo,
            boolean ignoreErrors) {
        final MetadataConverter[] metadataConverters =
                metadataKeys.stream()
                        .map(k -> Stream.of(ReadableMetadata.values())
                                .filter(rm -> rm.key.equals(k))
                                .findFirst()
                                .orElseThrow(IllegalStateException::new))
                        .map(m -> m.converter)
                        .toArray(MetadataConverter[]::new);

        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(inlongMetric)
                .withAuditAddress(auditHostAndPorts)
                .withAuditKeys(auditKeys)
                .build();

        final DynamicTubeMQDeserializationSchema<RowData> tubeMQDeserializer =
                new DynamicTubeMQTableDeserializationSchema(
                        deserialization, metadataConverters, producedTypeInfo, ignoreErrors,
                        innerFormat, metricOption, dirtySink, dirtyOptions);

        final FlinkTubeMQConsumer<RowData> tubeMQConsumer = new FlinkTubeMQConsumer(masterAddress, topic, streamIdSet,
                consumerGroup, tubeMQDeserializer, configuration, sessionKey, innerFormat);
        return tubeMQConsumer;
    }
Watermark strategy that is used to generate per-partition watermark.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/TubeMQTableSource.java