## Environment
* ClickHouse server version: 21.2.4 revision 54447

## Background
I m in the process of setting up a 2-node ClickHouse cluster, each node is with identical tables (as follows) for fetching data from Kafka topic.

```
CREATE TABLE `queue_05` (
  `id` UInt64,
  ...
)
ENGINE = Kafka()
SETTINGS
  kafka_broker_list = 'kafka01:9092,kafka02:9092,kafka03:9092',
  kafka_topic_list = 'topic03',
  kafka_group_name = 'g03',
  kafka_client_id = 'c03',
  kafka_format = 'JSONEachRow',
  kafka_num_consumers = 1
```

```
CREATE TABLE `table_03` (
  `id` UInt64,
  ...
)
ENGINE = ReplicatedReplacingMergeTree(
  '/clickhouse/tables/{shard}/table_03',
  '{replica}',
  updated_at
)
PARTITION BY toYYYYMM(`time`)
PRIMARY KEY id
```

```
CREATE MATERIALIZED VIEW `mv_03` TO `table_03`
  AS SELECT * FROM `queue_05`;
```

The Kafka topic `topic03` is with single partition. The ClickHouse tables `queue_05` in db01, db02 are in the same consumer group `g03` and are expected to work for failover purpose.

## Problem
When both tables `queue_05` in db01, db02 are attached, one of the table fetches data from single partition of topic `topic03`, the other table are in standby; when one node is down or one of the tables `queue_05` is detached, consumer group `g03` is then rebalancing and find no active members after a while, no failover happens.

Here is the log snippet of clickhouse-server.log in db02, right after I manually detach table `queue_05` in db01.
```
2021.02.24 11:43:46.110365 [ 1023 ] {} <Trace> StorageKafka (queue_05): Nothing to commit.
2021.02.24 11:43:46.110429 [ 1023 ] {} <Trace> StorageKafka (queue_05): Stream(s) stalled. Reschedule.
2021.02.24 11:43:46.610572 [ 1023 ] {} <Debug> StorageKafka (queue_05): Started streaming to 1 attached views
2021.02.24 11:43:46.611817 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
2021.02.24 11:43:46.611916 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
2021.02.24 11:43:46.611949 [ 1023 ] {} <Trace> StorageKafka (queue_05): Already subscribed to topics: [topic03]
2021.02.24 11:43:46.611959 [ 1023 ] {} <Trace> StorageKafka (queue_05): Already assigned to: [  ]
2021.02.24 11:43:46.611989 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
2021.02.24 11:43:46.662303 [ 1023 ] {} <Warning> StorageKafka (queue_05): Can't get assignment. It can be caused by some issue with consumer group (not enough partitions?). Will keep trying.
2021.02.24 11:43:46.662403 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_ASSIGNMENT in state up (join-state wait-assign-call)
2021.02.24 11:43:46.662465 [ 1023 ] {} <Trace> StorageKafka (queue_05): Nothing to commit.
2021.02.24 11:43:46.662582 [ 1023 ] {} <Trace> StorageKafka (queue_05): Stream(s) stalled. Reschedule.
2021.02.24 11:43:47.162758 [ 1023 ] {} <Debug> StorageKafka (queue_05): Started streaming to 1 attached views
2021.02.24 11:43:47.163558 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
2021.02.24 11:43:47.163614 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
2021.02.24 11:43:47.163642 [ 1023 ] {} <Trace> StorageKafka (queue_05): Already subscribed to topics: [topic03]
2021.02.24 11:43:47.163653 [ 1023 ] {} <Trace> StorageKafka (queue_05): Already assigned to: [  ]
2021.02.24 11:43:47.163680 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
2021.02.24 11:43:47.213860 [ 1023 ] {} <Warning> StorageKafka (queue_05): Can't get assignment. It can be caused by some issue with consumer group (not enough partitions?). Will keep trying.
2021.02.24 11:43:47.213967 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_ASSIGNMENT in state up (join-state wait-assign-call)
2021.02.24 11:43:47.214040 [ 1023 ] {} <Trace> StorageKafka (queue_05): Nothing to commit.
2021.02.24 11:43:47.214134 [ 1023 ] {} <Trace> StorageKafka (queue_05): Stream(s) stalled. Reschedule.
2021.02.24 11:43:47.714289 [ 1023 ] {} <Debug> StorageKafka (queue_05): Started streaming to 1 attached views
2021.02.24 11:43:47.715333 [ 1271 ] {} <Debug> StorageKafka (queue_05): [rdk:CGRPOP] [thrd:main]: Group "g03" received op GET_SUBSCRIPTION in state up (join-state wait-assign-call)
```

With the same Kafka broker settings, I setup another consumer group with 2 members (by `kafka-console-consumer.sh`) to consume data from single partition of topic `topic03`, when one member is down, consumer group rebalances and assigns the topic partition to the rest member successfully.


## Question
I guess I might have wrong configurations in ClickHouse and find no clue myself. Do you have idea of Kafka consumer group failover issue in my case? Any comment is appreciated.

inline bool hasMorePolledMessages() const
    {
        return (stalled_status == NOT_STALLED) && (current != messages.end());
    }
    inline bool polledDataUnusable() const
    {
        return  (stalled_status != NOT_STALLED) && (stalled_status != NO_MESSAGES_RETURNED);
    }
    inline bool isStalled() const { return stalled_status != NOT_STALLED; }
    void storeLastReadMessageOffset();
    void resetToLastCommitted(const char * msg);
    
    
    
    
    bool poll();
    
    String currentTopic() const { return current[-1].get_topic(); }
    String currentKey() const { return current[-1].get_key(); }
    auto currentOffset() const { return current[-1].get_offset(); }
    auto currentPartition() const { return current[-1].get_partition(); }
    auto currentTimestamp() const { return current[-1].get_timestamp(); }
    const auto & currentHeaderList() const { return current[-1].get_header_list(); }
    String currentPayload() const { return current[-1].get_payload(); }
private:
    using Messages = std::vector<cppkafka::Message>;
    enum StalledStatus
    {
        NOT_STALLED,
        NO_MESSAGES_RETURNED,