public void runFetchLoop() throws Exception {
        topicToThread = new HashMap<>();
        ExceptionProxy exceptionProxy = new ExceptionProxy(Thread.currentThread());

        try {

            while (running) {
                // re-throw any exception from the concurrent fetcher threads
                exceptionProxy.checkAndThrowException();

                // wait for max 5 seconds trying to get partitions to assign
                // if threads shut down, this poll returns earlier, because the threads inject the
                // special marker into the queue
                List<PulsarTopicState<T>> topicsToAssign = unassignedPartitionsQueue.getBatchBlocking(5000);
                // if there are more markers, remove them all
                topicsToAssign.removeIf(s -> s.equals(poisonState));

                if (!topicsToAssign.isEmpty()) {

                    if (!running) {
                        throw BreakingException.INSTANCE;
                    }

                    topicToThread.putAll(
                            createAndStartReaderThread(topicsToAssign, exceptionProxy));

                } else {
                    // there were no partitions to assign. Check if any consumer threads shut down.
                    // we get into this section of the code, if either the poll timed out, or the
                    // blocking poll was woken up by the marker element

                    topicToThread.values().removeIf(t -> !t.isRunning());

                }

                if (topicToThread.size() == 0 && unassignedPartitionsQueue.isEmpty()) {
                    PulsarTopicState topicForBlocking = unassignedPartitionsQueue.getElementBlocking();
                    if (topicForBlocking.equals(poisonState)) {
                        throw BreakingException.INSTANCE;
                    }
                    topicToThread.putAll(
                            createAndStartReaderThread(ImmutableList.of(topicForBlocking), exceptionProxy));
                }
            }

        } catch (BreakingException b) {
            // do nothing
        } catch (InterruptedException e) {
            // this may be thrown because an exception on one of the concurrent fetcher threads
            // woke this thread up. make sure we throw the root exception instead in that case
            exceptionProxy.checkAndThrowException();

            // no other root exception, throw the interrupted exception
            throw e;
        } finally {
            running = false;

            // clear the interruption flag
            // this allows the joining on reader threads (on best effort) to happen in
            // case the initial interrupt already
            Thread.interrupted();

            // make sure that in any case (completion, abort, error), all spawned threads are stopped
            try {
                int runningThreads = 0;
                do { // check whether threads are alive and cancel them
                    runningThreads = 0;

                    topicToThread.values().removeIf(s -> !s.isAlive());

                    for (ReaderThread t : topicToThread.values()) {
                        t.cancel();
                        runningThreads++;
                    }

                    if (runningThreads > 0) {
                        for (ReaderThread t : topicToThread.values()) {
                            t.join(500 / runningThreads + 1);
                        }

                    }

                } while (runningThreads > 0);

            } catch (InterruptedException ignored) {
                // waiting for the thread shutdown apparently got interrupted
                // restore interrupted state and continue
                Thread.currentThread().interrupt();
            } catch (Throwable t) {
                // we catch all here to preserve the original exception
                log.error("Exception while shutting down reader threads", t);
            }
        }
    }
The metric group which all metrics for the source should be registered to.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarFetcher.java