code: function topicsToPartitionAssignment(topicsInfo, topics, atTimestamp, defaultOffset = -1) {
      topics = _.isArray(topics) ? topics : topics.split(',');
  
      // This will be merged into the partition assignment, to either have the assignment
      // start from atTimestamp if given, else defaultOffset.
      const defaultResumePosition = atTimestamp ? {timestamp: atTimestamp} : {offset: defaultOffset};
  
      // Find the topic-partition metadata we want
      return _.flatten(
          topicsInfo.filter(t => _.includes(topics, t.name))
          // Map them into topic, partition, merging in the
          // defaultResumePosition of either timestamp or offset.
          .map(
              t => t.partitions.map(
                  p => Object.assign({topic: t.name, partition: p.id}, defaultResumePosition)
              )
          )
      );
  }
docs: Given topic names, this will create TopicPartition assignment objects
  for every partition in each topic, with either timestamp: atTimestamp
  or offset: defaultOffset.
  
  @param  {Array}  topicsInfo  topic metadata object, _metadata.topics.
  @param  {Array|String} of topics to build TopicPartition assignment objects for.
  @param  {int}  atTimestamp   If given, the assignment will have timestamp: atTimestamp set
  @param  {int}  defaultOffset If atTimestamp is not given, the assignment will have
  offset: defaultOffset set.  Default: -1 (latest offset)
  @return {Array} TopicPartition assignments, suitable for passing to buildAssignmentsAsync.
  If atTimestamp is given, the returned array SHOULD NOT be used to
  pass to KafkaConsumer.assign.  First resolve the timestamps into
  offsets by passing the assignments through buildAssignmentsAsync().