@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
        final LogicalType physicalType = physicalDataType.getLogicalType();
        final int physicalFieldCount = LogicalTypeChecks.getFieldCount(physicalType);
        final IntStream physicalFields = IntStream.range(0, physicalFieldCount);

        final DeserializationSchema<RowData> deserialization = createDeserialization(context,
                valueDecodingFormat, physicalFields.toArray(), null);

        final TypeInformation<RowData> producedTypeInfo = context.createTypeInformation(physicalDataType);

        final FlinkTubeMQConsumer<RowData> tubeMQConsumer = createTubeMQConsumer(deserialization, producedTypeInfo,
                ignoreErrors);

        return SourceFunctionProvider.of(tubeMQConsumer, false);
    }
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