@Override
        public void run() {
            StringBuilder sBuilder = new StringBuilder(256);
            final Long curThreadId = Thread.currentThread().getId();
            fetchWorkerStatusMap.put(curThreadId, 0);
            while (!isShutdown()) {
                PartitionSelectResult partSelectResult = null;
                fetchWorkerStatusMap.put(curThreadId, 0);
                try {
                    if (isShutdown()) {
                        break;
                    }
                    fetchWorkerStatusMap.put(curThreadId, 1);
                    MessageFetchManager.this.pushConsumer.allowConsumeWait();
                    partSelectResult =
                            MessageFetchManager.this.pushConsumer
                                    .getBaseConsumer().pushSelectPartition();
                    if (partSelectResult == null) {
                        continue;
                    }
                    Partition partition = partSelectResult.getPartition();
                    long usedToken = partSelectResult.getUsedToken();
                    boolean isLastConsumed = partSelectResult.isLastPackConsumed();
                    if (isShutdown()) {
                        MessageFetchManager.this.pushConsumer
                                .getBaseConsumer()
                                .pushReqReleasePartition(partition.getPartitionKey(), usedToken, isLastConsumed);
                        partSelectResult = null;
                        break;
                    }
                    if (MessageFetchManager.this.pushConsumer.isConsumePaused()) {
                        boolean result = partSelectResult.isLastPackConsumed();
                        if (result) {
                            result =
                                    MessageFetchManager.this.pushConsumer
                                            .getBaseConsumer().flushLastRequest(partition);
                        }
                        MessageFetchManager.this.pushConsumer
                                .getBaseConsumer().pushReqReleasePartition(partition.getPartitionKey(),
                                        usedToken, result);
                        partSelectResult = null;
                        continue;
                    }
                } catch (Throwable e) {
                    if (partSelectResult != null) {
                        MessageFetchManager.this.pushConsumer
                                .getBaseConsumer()
                                .pushReqReleasePartition(partSelectResult.getPartition().getPartitionKey(),
                                        partSelectResult.getUsedToken(), false);
                    }
                    sBuilder.delete(0, sBuilder.length());
                    logger.warn(sBuilder.append("Thread {} has been interrupted 3.")
                            .append(Thread.currentThread().getName()).toString());
                    sBuilder.delete(0, sBuilder.length());
                }
                fetchWorkerStatusMap.put(curThreadId, 2);
                if (partSelectResult != null) {
                    MessageFetchManager.this.pushConsumer.processRequest(
                            partSelectResult, sBuilder);
                }
            }
            fetchWorkerStatusMap.remove(curThreadId);
        }
Stop fetch worker threads

@param onlySetStatus  Whether to only set the state without stopping the operation
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageFetchManager.java