<commit_before>var args        = require('yargs').argv;
var kafka       = require('kafka-node');
var nodeCache   = require('node-cache');
var _           = require('lodash');
var config      = require('../config.js');
var cache       = require('./Cache.js').cache;
var logger      = require('../logger.js').logger;

var kafkaClient = new kafka.Client(process.env.ZOOKEEPER_CONNECT || config.zkConnect, 'lag-monitor');
var kafkaOffset = new kafka.Offset(kafkaClient);


var getCachedItems = function(cb) {
    cache.keys(function(err, keys){
        cache.mget(keys, function(err, items){
            cb(keys, items);
        });
    });
};


var getOffsetFetch = function(cb) {
    getCachedItems(function(keys, consumers) {
        logger.trace(consumers, 'cached consumers');

        var map = [];
        keys.forEach(function(key) {
            var consumer = consumers[key];
            if (!!consumer) {
                logger.trace(consumer, 'offset consumer to update is');
                map = map.concat(consumer.map(function(consumertopic){
                    return {
                        topic: consumertopic.topic,
                        partition: consumertopic.partition,
                        time: -1,
                        maxNum: 1 };
                }));
            }
        });
        logger.trace(map, 'mapped consumers');
        cb(map);
    });
};


var updateCachedItems = function(updates, cb) {
    logger.trace({updates: updates},'updates to be cached');
    if (!updates || updates.length === 0) {
        if (!!cb) {
            cb();
        }
        return;
    }

    getCachedItems(function(keys, consumers) {
        keys.forEach(function(key){
            var consumer = consumers[key];
            logger.trace({consumer: consumer, key: key}, 'update consumer offset');

            if (!!consumer) {
                consumer.forEach(function(consumerTopic){
                    // map the update to the consumer
                    var offset = updates[consumerTopic.topic][consumerTopic.partition];
                    logger.trace({topic: consumerTopic, offset: offset}, 'offset to update');
                    if (offset.length) {
                        consumerTopic.end = parseInt(offset[0]);
                        consumerTopic.lag = consumerTopic.end - consumerTopic.offset - 1;
                    }
                });
                //save the consumer back to the cache
                logger.trace(consumer, 'saving updated consumer');
                cache.set(key, consumer);
            }
        });

        if(cb) {
            cb();
        }
    });
};


var getTopicOffsets = function(cb) {
    logger.trace('get the topic offsets');
    getOffsetFetch(function(items){
        logger.trace(items, 'the offset fetch');
        kafkaOffset.fetch(items, function(err, data){<commit_msg>fix off by 1 error<commit_after>var args        = require('yargs').argv;
var kafka       = require('kafka-node');
var nodeCache   = require('node-cache');
var _           = require('lodash');
var config      = require('../config.js');
var cache       = require('./Cache.js').cache;
var logger      = require('../logger.js').logger;

var kafkaClient = new kafka.Client(process.env.ZOOKEEPER_CONNECT || config.zkConnect, 'lag-monitor');
var kafkaOffset = new kafka.Offset(kafkaClient);


var getCachedItems = function(cb) {
    cache.keys(function(err, keys){
        cache.mget(keys, function(err, items){
            cb(keys, items);
        });
    });
};


var getOffsetFetch = function(cb) {
    getCachedItems(function(keys, consumers) {
        logger.trace(consumers, 'cached consumers');

        var map = [];
        keys.forEach(function(key) {
            var consumer = consumers[key];
            if (!!consumer) {
                logger.trace(consumer, 'offset consumer to update is');
                map = map.concat(consumer.map(function(consumertopic){
                    return {
                        topic: consumertopic.topic,
                        partition: consumertopic.partition,
                        time: -1,
                        maxNum: 1 };
                }));
            }
        });
        logger.trace(map, 'mapped consumers');
        cb(map);
    });
};


var updateCachedItems = function(updates, cb) {
    logger.trace({updates: updates},'updates to be cached');
    if (!updates || updates.length === 0) {
        if (!!cb) {
            cb();
        }
        return;
    }

    getCachedItems(function(keys, consumers) {
        keys.forEach(function(key){
            var consumer = consumers[key];
            logger.trace({consumer: consumer, key: key}, 'update consumer offset');

            if (!!consumer) {
                consumer.forEach(function(consumerTopic){
                    // map the update to the consumer
                    var offset = updates[consumerTopic.topic][consumerTopic.partition];
                    logger.trace({topic: consumerTopic, offset: offset}, 'offset to update');
                    if (offset.length) {
                        consumerTopic.end = parseInt(offset[0]);
                        consumerTopic.lag = consumerTopic.end - consumerTopic.offset;
                    }
                });
                //save the consumer back to the cache
                logger.trace(consumer, 'saving updated consumer');
                cache.set(key, consumer);
            }
        });

        if(cb) {
            cb();
        }
    });
};


var getTopicOffsets = function(cb) {
    logger.trace('get the topic offsets');
    getOffsetFetch(function(items){
        logger.trace(items, 'the offset fetch');
        kafkaOffset.fetch(items, function(err, data){
{'fraction_non_alphanumeric': 0.1282498691327866, 'fraction_numerical': 0.0017448961786773686, 'mean_word_length': 1.7906523855890943, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5342047', 'n_tokens_mistral': 1553, 'n_tokens_neox': 1470, 'n_words': 330}