docs: def set_topic_partitions(self, *topics):
          """
          Set the topic/partitions to consume
          Optionally specify offsets to start from
  
          Accepts types:
  
          * str (utf-8): topic name (will consume all available partitions)
          * tuple: (topic, partition)
          * dict:
              - { topic: partition }
              - { topic: [partition list] }
              - { topic: (partition tuple,) }
  
          Optionally, offsets can be specified directly:
  
          * tuple: (topic, partition, offset)
          * dict:  { (topic, partition): offset, ... }
  
          Example:
  
          .. code:: python
  
              kafka = KafkaConsumer()
  
              # Consume topic1-all; topic2-partition2; topic3-partition0
              kafka.set_topic_partitions("topic1", ("topic2", 2), {"topic3": 0})
  
              # Consume topic1-0 starting at offset 12, and topic2-1 at offset 45
              # using tuples --
              kafka.set_topic_partitions(("topic1", 0, 12), ("topic2", 1, 45))
  
              # using dict --
              kafka.set_topic_partitions({ ("topic1", 0): 12, ("topic2", 1): 45 })
  
          """
          self._topics = []
          self._client.load_metadata_for_topics()
  
          # Setup offsets
          self._offsets = OffsetsStruct(fetch=dict(),
                                        commit=dict(),
                                        highwater=dict(),
                                        task_done=dict())
  
          # Handle different topic types
          for arg in topics:
  
              # Topic name str -- all partitions
              if isinstance(arg, (six.string_types, six.binary_type)):
                  topic = kafka_bytestring(arg)
  
                  for partition in self._client.get_partition_ids_for_topic(topic):
                      self._consume_topic_partition(topic, partition)
  
              # (topic, partition [, offset]) tuple
              elif isinstance(arg, tuple):
                  topic = kafka_bytestring(arg[0])
                  partition = arg[1]
                  self._consume_topic_partition(topic, partition)
                  if len(arg) == 3:
                      offset = arg[2]
                      self._offsets.fetch[(topic, partition)] = offset
  
              # { topic: partitions, ... } dict
              elif isinstance(arg, dict):
                  for key, value in six.iteritems(arg):
  
                      # key can be string (a topic)
                      if isinstance(key, (six.string_types, six.binary_type)):
                          topic = kafka_bytestring(key)
  
                          # topic: partition
                          if isinstance(value, int):
                              self._consume_topic_partition(topic, value)
  
                          # topic: [ partition1, partition2, ... ]
                          elif isinstance(value, (list, tuple)):
                              for partition in value:
                                  self._consume_topic_partition(topic, partition)
                          else:
                              raise KafkaConfigurationError(
                                  'Unknown topic type '
                                  '(dict key must be int or list/tuple of ints)'
                              )
  
                      # (topic, partition): offset
                      elif isinstance(key, tuple):
                          topic = kafka_bytestring(key[0])
                          partition = key[1]
                          self._consume_topic_partition(topic, partition)
                          self._offsets.fetch[(topic, partition)] = value
  
              else:
                  raise KafkaConfigurationError('Unknown topic type (%s)' % type(arg))
  
          # If we have a consumer group, try to fetch stored offsets
          if self._config['group_id']:
              self._get_commit_offsets()
  
          # Update missing fetch/commit offsets
          for topic_partition in self._topics:
  
              # Commit offsets default is None
              if topic_partition not in self._offsets.commit:
                  self._offsets.commit[topic_partition] = None
  
              # Skip if we already have a fetch offset from user args
              if topic_partition not in self._offsets.fetch:
  
                  # Fetch offsets default is (1) commit
                  if self._offsets.commit[topic_partition] is not None:
                      self._offsets.fetch[topic_partition] = self._offsets.commit[topic_partition]
  
                  # or (2) auto reset
                  else:
                      self._offsets.fetch[topic_partition] = \
                          self._reset_partition_offset(topic_partition)
  
          # highwater marks (received from server on fetch response)
          # and task_done (set locally by user)
          # should always get initialized to None
          self._reset_highwater_offsets()
          self._reset_task_done_offsets()
  
          # Reset message iterator in case we were in the middle of one
          self._reset_message_iterator()