@Override
    public Set<ConfigOption<?>> forwardOptions() {
        return Stream.of(
                TOPIC,
                ADMIN_URL,
                SERVICE_URL,
                SOURCE_SUBSCRIPTION_TYPE,
                SOURCE_SUBSCRIPTION_NAME,
                STARTUP_MODE,
                SOURCE_START_FROM_PUBLISH_TIME,
                SOURCE_STOP_AT_MESSAGE_ID,
                SOURCE_STOP_AFTER_MESSAGE_ID,
                SOURCE_STOP_AT_PUBLISH_TIME,
                SINK_CUSTOM_TOPIC_ROUTER,
                SINK_TOPIC_ROUTING_MODE,
                SINK_MESSAGE_DELAY_INTERVAL)
                .collect(Collectors.toSet());
    }
Format and Delivery guarantee related options are not forward options.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/PulsarTableFactory.java