@Override
        public void run() {
            boolean hasPermit;
            while (true) {
                hasPermit = false;
                long fetchTimeCost = -1;
                try {
                    if (context.getConfig().isStopConsume() || stopConsume) {
                        TimeUnit.MILLISECONDS.sleep(50L);
                        continue;
                    }

                    if (sleepTime > 0) {
                        TimeUnit.MILLISECONDS.sleep(sleepTime);
                    }

                    context.acquireRequestPermit();
                    hasPermit = true;
                    context.addConsumeTime(topic, -1);

                    long startFetchTime = System.currentTimeMillis();
                    ConsumerResult message = messageConsumer.getMessage();
                    fetchTimeCost = System.currentTimeMillis() - startFetchTime;
                    if (null != message && TErrCodeConstants.SUCCESS == message.getErrCode()) {
                        for (Message msg : message.getMessageList()) {
                            List<InLongMessage> msgs = new ArrayList<>();
                            List<InLongMessage> deserialize = deserializer
                                    .deserialize(context, topic, getAttributeMap(msg.getAttribute()),
                                            msg.getData());
                            context.addConsumeSuccess(topic, -1, deserialize.size(), msg.getData().length,
                                    fetchTimeCost);
                            int originSize = deserialize.size();
                            deserialize = interceptor.intercept(deserialize);
                            if (deserialize.isEmpty()) {
                                continue;
                            }
                            int filterSize = originSize - deserialize.size();
                            context.addConsumeFilter(topic, -1, filterSize);

                            msgs.addAll(deserialize);
                            handleAndCallbackMsg(new MessageRecord(topic.getTopicKey(), msgs,
                                    message.getConfirmContext(), System.currentTimeMillis()));
                        }
                        sleepTime = 0L;
                    } else {
                        context.addConsumeEmpty(topic, -1, fetchTimeCost);
                        emptyFetchTimes++;
                        if (emptyFetchTimes >= context.getConfig().getEmptyPollTimes()) {
                            sleepTime = Math.min((sleepTime += context.getConfig().getEmptyPollSleepStepMs()),
                                    context.getConfig().getMaxEmptyPollSleepMs());
                            emptyFetchTimes = 0;
                        }
                    }
                } catch (Exception e) {
                    context.addConsumeError(topic, -1, fetchTimeCost);
                    LOG.error(e.getMessage(), e);
                } finally {
                    if (hasPermit) {
                        context.releaseRequestPermit();
                    }
                }

                if (closed) {
                    break;
                }
            }
        }
parseAttr from k1=v1&k2=v2 to kv map

@param splitter {@link Splitter}
@param attr String
@param entrySplitterStr String
@return {@link Map}
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/tube/TubeSingleTopicFetcher.java