title: python kafka: how to make each msg consumed only once by group from begining
body: <p>I am using <a href="http://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html" rel="nofollow">Kafka consumer</a> here (version 1.3.1). </p>
  
  <p>What I am going to acheive: </p>
  
  <ul>
  <li><p>There are 10 partitions. each partition begins with offset 0. </p></li>
  <li><p>There is a group of consumers (1,2,3, eg). </p></li>
  <li><p>Sometimes, one consumer is down or is up. </p></li>
  <li><p>So, the group members may change. But I want each message in each partition should be consumed by the group only once (1 OR 2 OR 3).</p></li>
  </ul>
  
  <p>My codes are:</p>
  
  <pre><code>consumer = KafkaConsumer('my_topic',
              bootstrap_servers=['ip:9092'],
              auto_offset_reset='earliest',
              max_partition_fetch_bytes=131072,
              group_id='writer.test')
  </code></pre>
  
  <p>Is the above configuration enough? Any comments welcomed. Thanks</p>
  
  <p><strong>UPDATE</strong></p>
  
  <p><strong>I tried the following codes. Each time in partition 760, each message maybe consumed twice by two consumers in one group. Why? Something wrong?</strong></p>
  
  <pre><code>def test():
      #PULL FROM KAFKA
      consumer = KafkaConsumer(
              'topic',
              bootstrap_servers=[ip],
              auto_offset_reset='latest',
              max_partition_fetch_bytes=131072,
              auto_commit_interval_ms=500,
              group_id='writer2.test')
  
      print consumer.poll()
      for i in range(10000):
          msg = next(consumer)
          if str(msg[1])=='670':
              print 'partition= %s, offset= %s' % (msg[1], msg[2])
      consumer.unsubscribe()
  
  
  if __name__ == "__main__":
      for i in range(10):
          import time
          time.sleep(5)
          test()
  </code></pre>
  
  <p>Output 1:</p>
  
  <pre><code>{}
  partition= 670, offset= 224
  partition= 670, offset= 225
  partition= 670, offset= 226
  partition= 670, offset= 227
  partition= 670, offset= 228
  partition= 670, offset= 229
  partition= 670, offset= 230
  partition= 670, offset= 231
  partition= 670, offset= 232
  partition= 670, offset= 233
  partition= 670, offset= 234
  partition= 670, offset= 235
  partition= 670, offset= 236
  partition= 670, offset= 237
  partition= 670, offset= 238
  partition= 670, offset= 239
  partition= 670, offset= 240
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  </code></pre>
  
  <p>Run the same file in another window, output:</p>
  
  <pre><code>{}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  {}
  partition= 670, offset= 241
  partition= 670, offset= 242
  partition= 670, offset= 243
  partition= 670, offset= 244
  partition= 670, offset= 245
  partition= 670, offset= 246
  partition= 670, offset= 247
  partition= 670, offset= 248
  partition= 670, offset= 249
  partition= 670, offset= 250
  partition= 670, offset= 251
  partition= 670, offset= 252
  partition= 670, offset= 253
  partition= 670, offset= 254
  partition= 670, offset= 255
  partition= 670, offset= 256
  partition= 670, offset= 257
  partition= 670, offset= 258
  partition= 670, offset= 259
  </code></pre>