private static Optional<DecodingFormat<DeserializationSchema<RowData>>> getKeyDecodingFormat(
            TableFactoryHelper helper) {
        final Optional<DecodingFormat<DeserializationSchema<RowData>>> keyDecodingFormat = helper
                .discoverOptionalDecodingFormat(DeserializationFormatFactory.class, KEY_FORMAT);
        keyDecodingFormat.ifPresent(format -> {
            if (!format.getChangelogMode().containsOnly(RowKind.INSERT)) {
                throw new ValidationException(String.format(
                        "A key format should only deal with INSERT-only records. "
                                + "But %s has a changelog mode of %s.",
                        helper.getOptions().get(KEY_FORMAT),
                        format.getChangelogMode()));
            }
        });
        return keyDecodingFormat;
    }
A dynamic table factory implementation for TubeMQ.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/TubeMQDynamicTableFactory.java