@Override
    public void record(
            Map<String, ?> source, Map<String, ?> position, String databaseName, String ddl)
            throws DatabaseHistoryException {
        throw new UnsupportedOperationException(
                String.format(
                        "The %s cannot work with 'debezium.internal.implementation' = 'legacy',"
                                + "please use %s",
                        FlinkDatabaseSchemaHistory.class.getCanonicalName(),
                        FlinkDatabaseHistory.class.getCanonicalName()));
    }
Determine whether the {@link FlinkDatabaseSchemaHistory} is compatible with the specified
state.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/FlinkDatabaseSchemaHistory.java