@Override
    public void cancel() {
        // set ourselves as not running;
        // this would let the main discovery loop escape as soon as possible
        running = false;

        if (discoveryLoopThread != null) {

            if (partitionDiscoverer != null) {
                // we cannot close the discoverer here, as it is error-prone to concurrent access;
                // only wakeup the discoverer, the discovery loop will clean itself up after it
                // escapes
                partitionDiscoverer.wakeup();
            }

            // the discovery loop may currently be sleeping in-between
            // consecutive discoveries; interrupt to shutdown faster
            discoveryLoopThread.interrupt();
        }

        // abort the fetcher, if there is one
        if (kafkaFetcher != null) {
            kafkaFetcher.cancel();
        }
    }
By default, when restoring from a checkpoint / savepoint, the consumer always ignores
restored partitions that are no longer associated with the current specified topics or topic
pattern to subscribe to.

<p>This method configures the consumer to not filter the restored partitions, therefore
always attempting to consume whatever partition was present in the previous execution
regardless of the specified topics to subscribe to in the current execution.

@return The consumer object, to allow function chaining.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaConsumerBase.java