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();  

        });  

（编辑：辽源站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!