code: func (s *KafkaSource) Generate(f *flow.Flow) *flow.Dataset {
  	partitionIds, err := s.fetchPartitionIds()
  	if err != nil {
  		log.Printf("KafkaSource failed to fetch kafka partitions: %v", err)
  		return nil
  	}
  	return s.genShardInfos(f, partitionIds).
  		RoundRobin(s.prefix, len(partitionIds)).
  		Map(s.prefix+".Read", MapperReadShard)
  }
docs: // Generate generates data shard info,
  // partitions them via round robin,
  // and reads each shard on each executor