def set_topic_partitions(self, *topics):
self._topics = []<EOL>self._client.load_metadata_for_topics()<EOL>self._offsets = OffsetsStruct(fetch=dict(),<EOL>commit=dict(),<EOL>highwater=dict(),<EOL>task_done=dict())<EOL>for arg in topics:<EOL><INDENT>if isinstance(arg, (six.string_types, six.binary_type)):<EOL><INDENT>topic = kafka_bytestring(arg)<EOL>for partition in self._client.get_partition_ids_for_topic(topic):<EOL><INDENT>self._consume_topic_partition(topic, partition)<EOL><DEDENT><DEDENT>elif isinstance(arg, tuple):<EOL><INDENT>topic = kafka_bytestring(arg[<NUM_LIT:0>])<EOL>partition = arg[<NUM_LIT:1>]<EOL>self._consume_topic_partition(topic, partition)<EOL>if len(arg) == <NUM_LIT:3>:<EOL><INDENT>offset = arg[<NUM_LIT:2>]<EOL>self._offsets.fetch[(topic, partition)] = offset<EOL><DEDENT><DEDENT>elif isinstance(arg, dict):<EOL><INDENT>for key, value in six.iteritems(arg):<EOL><INDENT>if isinstance(key, (six.string_types, six.binary_type)):<EOL><INDENT>topic = kafka_bytestring(key)<EOL>if isinstance(value, int):<EOL><INDENT>self._consume_topic_partition(topic, value)<EOL><DEDENT>elif isinstance(value, (list, tuple)):<EOL><INDENT>for partition in value:<EOL><INDENT>self._consume_topic_partition(topic, partition)<EOL><DEDENT><DEDENT>else:<EOL><INDENT>raise KafkaConfigurationError(<EOL>'<STR_LIT>'<EOL>'<STR_LIT>'<EOL>)<EOL><DEDENT><DEDENT>elif isinstance(key, tuple):<EOL><INDENT>topic = kafka_bytestring(key[<NUM_LIT:0>])<EOL>partition = key[<NUM_LIT:1>]<EOL>self._consume_topic_partition(topic, partition)<EOL>self._offsets.fetch[(topic, partition)] = value<EOL><DEDENT><DEDENT><DEDENT>else:<EOL><INDENT>raise KafkaConfigurationError('<STR_LIT>' % type(arg))<EOL><DEDENT><DEDENT>if self._config['<STR_LIT>']:<EOL><INDENT>self._get_commit_offsets()<EOL><DEDENT>for topic_partition in self._topics:<EOL><INDENT>if topic_partition not in self._offsets.commit:<EOL><INDENT>self._offsets.commit[topic_partition] = None<EOL><DEDENT>if topic_partition not in self._offsets.fetch:<EOL><INDENT>if self._offsets.commit[topic_partition] is not None:<EOL><INDENT>self._offsets.fetch[topic_partition] = self._offsets.commit[topic_partition]<EOL><DEDENT>else:<EOL><INDENT>self._offsets.fetch[topic_partition] =self._reset_partition_offset(topic_partition)<EOL><DEDENT><DEDENT><DEDENT>self._reset_highwater_offsets()<EOL>self._reset_task_done_offsets()<EOL>self._reset_message_iterator()<EOL>
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 })