private Class<?> determineDatabase() {
        boolean isCompatibleWithLegacy =
                FlinkDatabaseHistory.isCompatible(retrieveHistory(engineInstanceName));
        if (LEGACY_IMPLEMENTATION_VALUE.equals(properties.get(LEGACY_IMPLEMENTATION_KEY))) {
            // specifies the legacy implementation but the state may be incompatible
            if (isCompatibleWithLegacy) {
                return FlinkDatabaseHistory.class;
            } else {
                throw new IllegalStateException(
                        "The configured option 'debezium.internal.implementation' is 'legacy', but the state of source is incompatible with this implementation, you should remove the the option.");
            }
        } else if (FlinkDatabaseSchemaHistory.isCompatible(retrieveHistory(engineInstanceName))) {
            // tries the non-legacy first
            return FlinkDatabaseSchemaHistory.class;
        } else if (isCompatibleWithLegacy) {
            // fallback to legacy if possible
            return FlinkDatabaseHistory.class;
        } else {
            // impossible
            throw new IllegalStateException("Can't determine which DatabaseHistory to use.");
        }
    }
Safely and gracefully stop the Debezium engine.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/DebeziumSourceFunction.java