Kafka之消费与激情

        // We do not want to be stuck blocking in poll if we are missing some positions  

        // since the offset lookup may be backing off after a failure  

        // NOTE: the use of cachedSubscriptionHashAllFetchPositions means we MUST call  

        // updateAssignmentMetadataIfNeeded before this method.  

        if (!cachedSubscriptionHashAllFetchPositions && pollTimeout > retryBackoffMs) {  

            pollTimeout = retryBackoffMs;  

        }  

        client.poll(pollTimeout, startMs, () -> {  

            // since a fetch might be completed by the background thread, we need this poll condition  

            // to ensure that we do not block unnecessarily in poll()  

            return !fetcher.hasCompletedFetches();  

        });  

（编辑：ASP站长）

【免责声明】本站内容转载自互联网，其相关言论仅代表作者个人观点绝非权威，不代表本站立场。如您发现内容存在版权问题，请提交相关链接至邮箱：bqsm@foxmail.com，我们将及时予以处理。