lib/utils.js
function assignmentsForTimesAsync(kafkaConsumer, assignments) {
    const assignmentsWithTimestamps = assignments.filter(a => 'timestamp' in a && !('offset' in a)).map((a) => {
        return {topic: a.topic, partition: a.partition, offset: a.timestamp };
    });

    // If there were no timestamps to resolve, just return assignments as is.
    if (assignmentsWithTimestamps.length == 0) {
        return new P((resolve, reject) => resolve(assignments));
    }
    // Else resolve all timestamp assignments to offsets.
    else {
        // Get offsets for the timestamp based assignments.
        // If no offset was found for any timestamp, this will return -1, which
        // we can use for offset: -1 as end of partition.
        return kafkaConsumer.offsetsForTimesAsync(assignmentsWithTimestamps)
        .then(offsetsForTimes => {
            // console.log({ withTimestamps: assignmentsWithTimestamps, got: offsetsForTimes}, 'got offsets for times');
            // merge offsetsForTimes with any non-timestamp based assignments.
            // This merged object will only have assignments with offsets specified, no timestamps.
            return _.flatten([
                assignments.filter(a => !('timestamp' in a) && 'offset' in a),
                offsetsForTimes
            ]);
        })
    }
}
function assignmentsForTimesAsync(kafkaConsumer, assignments) {
    const assignmentsWithTimestamps = assignments.filter(a => 'timestamp' in a && !('offset' in a)).map((a) => {
        return {topic: a.topic, partition: a.partition, offset: a.timestamp };
    });

    // If there were no timestamps to resolve, just return assignments as is.
    if (assignmentsWithTimestamps.length == 0) {
        return new P((resolve, reject) => resolve(assignments));
    }
    // Else resolve all timestamp assignments to offsets.
    else {
        // Get offsets for the timestamp based assignments.
        // If no offset was found for any timestamp, this will return -1, which
        // we can use for offset: -1 as end of partition.
        return kafkaConsumer.offsetsForTimesAsync(assignmentsWithTimestamps)
        .then(offsetsForTimes => {
            // console.log({ withTimestamps: assignmentsWithTimestamps, got: offsetsForTimes}, 'got offsets for times');
            // merge offsetsForTimes with any non-timestamp based assignments.
            // This merged object will only have assignments with offsets specified, no timestamps.
            return _.flatten([
                assignments.filter(a => !('timestamp' in a) && 'offset' in a),
                offsetsForTimes
            ]);
        })
    }
}
function assignmentsForTimesAsync ( kafkaConsumer , assignments ) { const assignmentsWithTimestamps = assignments . filter ( a => 'timestamp' in a && ! ( 'offset' in a ) ) . map ( ( a ) => { return { topic : a . topic , partition : a . partition , offset : a . timestamp } ; } ) ; // If there were no timestamps to resolve, just return assignments as is. if ( assignmentsWithTimestamps . length == 0 ) { return new P ( ( resolve , reject ) => resolve ( assignments ) ) ; } // Else resolve all timestamp assignments to offsets. else { // Get offsets for the timestamp based assignments. // If no offset was found for any timestamp, this will return -1, which // we can use for offset: -1 as end of partition. return kafkaConsumer . offsetsForTimesAsync ( assignmentsWithTimestamps ) . then ( offsetsForTimes => { // console.log({ withTimestamps: assignmentsWithTimestamps, got: offsetsForTimes}, 'got offsets for times'); // merge offsetsForTimes with any non-timestamp based assignments. // This merged object will only have assignments with offsets specified, no timestamps. return _ . flatten ( [ assignments . filter ( a => ! ( 'timestamp' in a ) && 'offset' in a ) , offsetsForTimes ] ) ; } ) } }
Given a a Kafka assignemnts array, this will look for any occurances
of 'timestamp' instead of 'offset'.  For those found, it will issue
a offsetsForTimes request on the kafkaConsumer.  The returned
offsets will be merged with any provided non-timestamp based assignments,
and returned as a Promise of assignments array with only offsets.

If no timestamps based assignments are given, this will just return a Promise
of the provided assignments.

@param  {KafkaConsumer} kafkaConsumer
@param  {Array}         assignments
@return {Promise}       Array of timestamp resolved assignment objects with offsets.
Given a a Kafka assignemnts array this will look for any occurances of timestamp instead of offset . For those found it will issue a offsetsForTimes request on the kafkaConsumer . The returned offsets will be merged with any provided non - timestamp based assignments and returned as a Promise of assignments array with only offsets .