kafka手动管理offset，poll 请求拉取失败，一直轮询 ？

多实例，手动管理offset，存储到redis，项目启动初始化时，读取offset，使用seek订阅消费，poll 拉取消息，刚调试的时候正常，过几天看下，poll 拉取消息失败

Kafka version : 0.10.1.1 , 多实例环境，手动管理offset

多线程消费，每个线程消费固定的 partition

 @Override
    public void run() {
        consumer.assign(Arrays.asList(topicPartition));
        String key = StartInit.REDIS_KEY_MSG + "topic:" + topicPartition.topic() + "partition:" + topicPartition.partition();
        String lockKey = REDIS_LOCK_KEY + "topic:" + topicPartition.topic() + "partition:" + topicPartition.partition();
        long position = getOffset(key);
        log.info("topic [{}], partition [{}] thread start,consume begin offset is [{}]  ...", topicPartition.topic(), topicPartition.partition(), position);
        consumer.seek(topicPartition, position);
        try {
            boolean running = true;
            while (running) {
                long offsetNewest;
                try {
                    ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                    //保证事务
                    CustomTransactionCallback customTransactionCallback = new CustomTransactionCallback(0L, records);
                    transactionTemplate.execute(customTransactionCallback);
                    offsetNewest = customTransactionCallback.getOffsetNewest();
                } catch (Exception e) {
                    log.error("find exception [{}] ...", e);
                    continue;
                }
                consumer.commitSync();
                log.info("topic [{}], partition [{}], offsetNewest is [{}] ...", topicPartition.topic(), topicPartition.partition(), offsetNewest);
                boolean locked = redisManager.lock(lockKey, 180, 3);
                if (locked) {
                    try {
                        long offsetDb = getOffset(key);
                        if (offsetDb < offsetNewest) {
                            // 更新 redis
                            stringRedisTemplate.opsForValue().set(key, String.valueOf(offsetNewest));
                        }
                    } finally {
                        redisLockService.unlock(lockKey);
                    }
                }
            }
        } finally {
            consumer.close();
        }

    }

1. 2019-12-13 09:34:24.613 [kafka_msg_consumer_2] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-0
   2019-12-13 09:34:28.034 [kafka_msg_consumer_1] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-1
   2019-12-13 09:34:28.066 [kafka_msg_consumer_1] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-1
   2019-12-13 09:34:28.081 [kafka_msg_consumer_0] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-2
   2019-12-13 09:34:28.081 [kafka_msg_consumer_1] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-1
   2019-12-13 09:34:28.096 [kafka_msg_consumer_2] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-0
   2019-12-13 09:34:28.113 [kafka_msg_consumer_1] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-1
   2019-12-13 09:34:28.113 [kafka_msg_consumer_0] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-2
   2019-12-13 09:34:28.144 [kafka_msg_consumer_1] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-1
   2019-12-13 09:34:28.144 [kafka_msg_consumer_0] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-2
   2019-12-13 09:34:28.159 [kafka_msg_consumer_2] WARN  o.apache.kafka.clients.consumer.internals.Fetcher - Unknown error fetching data for topic-partition topic-mescenter-mes-0
   .....
   

   谢谢解答。