## 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.

return false;
                }
            }
            else
            {
                LOG_TRACE(log, "Stalled");
                return false;
            }
        }
        else
        {
            messages = std::move(new_messages);
            current = messages.begin();
            LOG_TRACE(log, "Polled batch of {} messages. Offsets position: {}",
                messages.size(), consumer->get_offsets_position(consumer->get_assignment()));
            break;
        }
    }
    filterMessageErrors();
    if (current == messages.end())
    {
        LOG_ERROR(log, "Only errors left");
        stalled_status = ERRORS_RETURNED;
        return false;
    }
    stalled_status = NOT_STALLED;
    allowed = true;
    return true;
}
size_t ReadBufferFromKafkaConsumer::filterMessageErrors()