kafka的listeners配置错误导致主线程阻塞

public KafkaConsumerImpl init() {
    if (group == null || group.isEmpty()) {
        throw new RuntimeException("phoenix.mq.group is empty");
    }
    Properties props = new Properties();
    
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, namesrvAddr);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
    // 是否允许自动提交offset，这里设为false，下面手动提交
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    // ...
    consumer = new KafkaConsumer<>(props);

    // consumer 订阅的topic及partition
    topicPartition = new TopicPartition(this.topic, this.partitionId);
    this.partitions = Collections.singletonList(topicPartition);

    // 元数据初始化和连接测试，3次失败后抛出异常
    Callable<Boolean> call = new Callable<Boolean>() {
        boolean res = false;
        int tryTimes = 3;
        
        @Override
        public Boolean call() throws Exception {
            while (tryTimes-- > 0) {
                try {
                    consumer.assign(partitions);
                    // 默认初始化offset当前最大值
                    nextBeginOffset = consumer.position(topicPartition);
                    res = true;
                    break;
                } catch (Exception e) {
                    if (e instanceof InterruptedException) {
                        break; // 如果position在阻塞状态时，调用了 task.cancel 会抛出此异常。直接退出即可
                    }
                    LOG.error(e.getMessage(), e);
                    LOG.error(" ==> error when trying to fetch metadata for kafka. topic<{}>, partition<{}>", topic,
                            partitionId);
                }
                // sleep
                try {
                    Thread.sleep(2000);
                }
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return res;
        }
    };

    FutureTask<Boolean> task = new FutureTask<>(call);
    new Thread(task).start();

    boolean isOk = false;
    try {
        isOk = task.get(10000, TimeUnit.MILLISECONDS);
    }
    catch (Exception e) {
        LOG.error("Get task result timeout", e);
    }

    task.cancel(true);

    if (isOk) {
        LOG.info(" ==> init kafka consumer succeed: servers<{}>, topic<{}>, partition<{}>, nextBeginOffset<{}>",
                namesrvAddr, topic, partitionId, nextBeginOffset);
    }
    else {
        throw new RuntimeException(String.format(
                " ==> init kafka consumer failed. please check the conf (listeners or advertised.listeners or ...) and try to ping the host name in the conf value"));
    }
    return this;
}