src/healthCheckers/consumerOffsetOutOfSyncChecker.js
function isOffsetsInSync(notIncrementedTopicPayloads, zookeeperOffsets, kafkaOffsetDiffThreshold, logger) {
    logger.trace('Monitor Offset: Topics offsets', zookeeperOffsets);
    let lastErrorToHealthCheck;
    notIncrementedTopicPayloads.forEach(function (topicPayload) {
        let {topic, partition, offset} = topicPayload;
        if (zookeeperOffsets && zookeeperOffsets[topic] && zookeeperOffsets[topic][partition]) {
            let zkLatestOffset = zookeeperOffsets[topic][partition][0];
            let unhandledMessages = zkLatestOffset - offset;
            if (unhandledMessages > kafkaOffsetDiffThreshold) {
                let state = {
                    topic: topic,
                    partition: partition,
                    partitionLatestOffset: zkLatestOffset,
                    partitionReadOffset: offset,
                    unhandledMessages: unhandledMessages
                };

                logger.error('Monitor Offset: Kafka consumer offsets found to be out of sync', state);
                lastErrorToHealthCheck = new Error('Monitor Offset: Kafka consumer offsets found to be out of sync:' + JSON.stringify(state));
            }
        } else {
            logger.error('Monitor Offset: Kafka consumer topics/partitions found to be out of sync in topic: ' + topic + ' and in partition:' + partition);
            lastErrorToHealthCheck = new Error('Monitor Offset: Kafka consumer topics/partitions found to be out of sync in topic: ' + topic + ' and in partition:' + partition);
        }
    });

    return lastErrorToHealthCheck;
}
function isOffsetsInSync(notIncrementedTopicPayloads, zookeeperOffsets, kafkaOffsetDiffThreshold, logger) {
    logger.trace('Monitor Offset: Topics offsets', zookeeperOffsets);
    let lastErrorToHealthCheck;
    notIncrementedTopicPayloads.forEach(function (topicPayload) {
        let {topic, partition, offset} = topicPayload;
        if (zookeeperOffsets && zookeeperOffsets[topic] && zookeeperOffsets[topic][partition]) {
            let zkLatestOffset = zookeeperOffsets[topic][partition][0];
            let unhandledMessages = zkLatestOffset - offset;
            if (unhandledMessages > kafkaOffsetDiffThreshold) {
                let state = {
                    topic: topic,
                    partition: partition,
                    partitionLatestOffset: zkLatestOffset,
                    partitionReadOffset: offset,
                    unhandledMessages: unhandledMessages
                };

                logger.error('Monitor Offset: Kafka consumer offsets found to be out of sync', state);
                lastErrorToHealthCheck = new Error('Monitor Offset: Kafka consumer offsets found to be out of sync:' + JSON.stringify(state));
            }
        } else {
            logger.error('Monitor Offset: Kafka consumer topics/partitions found to be out of sync in topic: ' + topic + ' and in partition:' + partition);
            lastErrorToHealthCheck = new Error('Monitor Offset: Kafka consumer topics/partitions found to be out of sync in topic: ' + topic + ' and in partition:' + partition);
        }
    });

    return lastErrorToHealthCheck;
}
function isOffsetsInSync ( notIncrementedTopicPayloads , zookeeperOffsets , kafkaOffsetDiffThreshold , logger ) { logger . trace ( 'Monitor Offset: Topics offsets' , zookeeperOffsets ) ; let lastErrorToHealthCheck ; notIncrementedTopicPayloads . forEach ( function ( topicPayload ) { let { topic , partition , offset } = topicPayload ; if ( zookeeperOffsets && zookeeperOffsets [ topic ] && zookeeperOffsets [ topic ] [ partition ] ) { let zkLatestOffset = zookeeperOffsets [ topic ] [ partition ] [ 0 ] ; let unhandledMessages = zkLatestOffset - offset ; if ( unhandledMessages > kafkaOffsetDiffThreshold ) { let state = { topic : topic , partition : partition , partitionLatestOffset : zkLatestOffset , partitionReadOffset : offset , unhandledMessages : unhandledMessages } ; logger . error ( 'Monitor Offset: Kafka consumer offsets found to be out of sync' , state ) ; lastErrorToHealthCheck = new Error ( 'Monitor Offset: Kafka consumer offsets found to be out of sync:' + JSON . stringify ( state ) ) ; } } else { logger . error ( 'Monitor Offset: Kafka consumer topics/partitions found to be out of sync in topic: ' + topic + ' and in partition:' + partition ) ; lastErrorToHealthCheck = new Error ( 'Monitor Offset: Kafka consumer topics/partitions found to be out of sync in topic: ' + topic + ' and in partition:' + partition ) ; } } ) ; return lastErrorToHealthCheck ; }
Compares the consumer offsets vs ZooKeeper's offset Will return false if founds a diff
Compares the consumer offsets vs ZooKeeper s offset Will return false if founds a diff