Attempt to heart beat failed since the group is rebalancing, try to re-join group

In my project, I used @KakfaListener configure kafka containerFactory and topic.

Topic Name:

public static final String CONNECT_DEVICE_MESSAGE_TOPIC = "connectDeviceMessageTopic";

Topic Listen:

@KafkaListener(containerFactory = "receiveKafkaListenerContainerFactory", topics = KafkaQueueName.CONNECT_DEVICE_MESSAGE_TOPIC)
public void onMessageListener(MessageTemplate message){
}

Kafka Config:

package me.hekr.bot.parse.core.kafka;

import lombok.extern.slf4j.Slf4j;
import me.hekr.bot.utils.IpUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Neon Wang on 2016/10/20.
 */
@EnableKafka
@Configuration
@Slf4j
public class KafkaConfig {
    @Value("${bot.kafka.servers}")
    private String servers;

    /*********************** Producer Config ***************************/
    private ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    private Map<String, Object> producerConfigs() {
        return new CustomHashMap().put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers)
                    .put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
                    .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
                    .put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000 * 2);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    /*********************** Consumer Config ***************************/
    private Map<String, Object> consumerProps() {
        return new CustomHashMap()
                .put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers)
                .put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true)
                .put(ConsumerConfig.GROUP_ID_CONFIG, "parseReceiveMessageFormConnection")
                .put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100")
                .put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000")
                .put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
                .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
    receiveKafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.setMessageConverter(new StringJsonMessageConverter());
        factory.getContainerProperties().setPollTimeout(3000L);
        return factory;
    }

    private ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProps());
    }

    class CustomHashMap extends HashMap<String, Object> {

        CustomHashMap(){
            super();
        }

        @Override
        public CustomHashMap put(String key, Object value) {
            super.put(key, value);
            return this;
        }
    }
}

Started project, kafka configuration information was successed, but I found every topic information had logged three times, is it normal？

2017-08-16T11:12:43.633+0800 INFO  [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-kafka-consumer-1] o.a.k.c.c.i.AbstractCoordinator.handle:623 - Attempt to heart beat failed since the group is rebalancing, try to re-join group.

Then the first message had received again and again, all result turn out was same！

After three times, hadn't received any message, I had no idea, who can help me?

The group id had changed twice， in last version had used localhost ip.

IpUtil.getLocalhostAddress().replace(".", "")