@Override
            public void run() {
                StringBuilder strBuffer = new StringBuilder(256);
                while ((!isRebalanceStopped()) || (!isShutdown())) {
                    try {
                        ConsumerEvent event = rebalanceEvents.take();
                        rebalanceEvents.clear();
                        if ((isRebalanceStopped()) || (isShutdown())) {
                            break;
                        }
                        switch (event.getType()) {
                            case DISCONNECT:
                            case ONLY_DISCONNECT:
                                disconnectFromBroker(event);
                                rebalanceResults.put(event);
                                break;
                            case CONNECT:
                            case ONLY_CONNECT:
                                connect2Broker(event);
                                rebalanceResults.put(event);
                                break;
                            case REPORT:
                                reportSubscribeInfo();
                                break;
                            case STOPREBALANCE:
                                break;
                            default:
                                throw new TubeClientException(strBuffer
                                        .append("Invalid rebalance opCode:")
                                        .append(event.getType()).toString());
                        }
                        rebalanceRetryTimes = 0;
                    } catch (InterruptedException e) {
                        return;
                    } catch (Throwable e) {
                        rebalanceRetryTimes++;
                        if (!isShutdown()) {
                            strBuffer.delete(0, strBuffer.length());
                            logger.error(strBuffer.append("Rebalance retry ")
                                    .append(rebalanceRetryTimes).append(" failed.").toString(), e);
                            strBuffer.delete(0, strBuffer.length());
                        }
                    }
                }
            }
Construct a BaseMessageConsumer object.

@param sessionFactory session factory
@param consumerConfig consumer configuration
@param isPullConsume  if this is a pull consumer
@throws TubeClientException
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java