src/healthCheckers/consumerOffsetOutOfSyncChecker.js
function getNotIncrementedTopicPayloads(previousConsumerReadOffset, consumer) {
    let notIncrementedTopicPayloads = consumer.topicPayloads.filter((topicPayload) => {
        let {topic, partition, offset: currentOffset} = topicPayload;
        let previousTopicPayloadForPartition = _.find(previousConsumerReadOffset, {topic, partition});
        return previousTopicPayloadForPartition && currentOffset === previousTopicPayloadForPartition.offset;
    });

    return notIncrementedTopicPayloads;
}
function getNotIncrementedTopicPayloads(previousConsumerReadOffset, consumer) {
    let notIncrementedTopicPayloads = consumer.topicPayloads.filter((topicPayload) => {
        let {topic, partition, offset: currentOffset} = topicPayload;
        let previousTopicPayloadForPartition = _.find(previousConsumerReadOffset, {topic, partition});
        return previousTopicPayloadForPartition && currentOffset === previousTopicPayloadForPartition.offset;
    });

    return notIncrementedTopicPayloads;
}
function getNotIncrementedTopicPayloads ( previousConsumerReadOffset , consumer ) { let notIncrementedTopicPayloads = consumer . topicPayloads . filter ( ( topicPayload ) => { let { topic , partition , offset : currentOffset } = topicPayload ; let previousTopicPayloadForPartition = _ . find ( previousConsumerReadOffset , { topic , partition } ) ; return previousTopicPayloadForPartition && currentOffset === previousTopicPayloadForPartition . offset ; } ) ; return notIncrementedTopicPayloads ; }
Compares the current consumer offsets vs the its previous state Returns topic payload that wasn't incremented from last check
Compares the current consumer offsets vs the its previous state Returns topic payload that wasn t incremented from last check