如何使用 kafka.consumer.SimpleConsumer,seek()答案

【问题标题】：How to use kafka.consumer.SimpleConsumer,seek()如何使用 kafka.consumer.SimpleConsumer,seek()
【发布时间】：2015-10-28 18:53:24
【问题描述】：

API 文档在这里：http://kafka-python.readthedocs.org/en/latest/apidoc/kafka.consumer.html

但是当我运行下面的代码时，异常是%d format: a number is required, not NoneType

    client = KafkaClient("localhost:9092")
    consumer = SimpleConsumer(client, "test-group", "test")
    consumer.seek(0, whence=None)# (0,2) and (0,0)
    run = True
    while( run ):
        message = consumer.get_message(block=False, timeout=4000)

    except Exception as e:
        print "Exception while trying to read msg:", str(e)

当我使用下面这段代码时，异常是 seek() got an unexpected keyword argument 'partition'

consumer.seek(0, whence=None, partition=None)# (0,2) and (0,0)

有什么想法吗？谢谢。

【问题讨论】：

标签： python apache-kafka kafka-consumer-api kafka-python

【解决方案1】：

在 Kafka Definitive Guide 中，有一个用 Java 编写的示例代码 seek()（不是 Python，但我希望你能大致了解）。

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {

         public void onPartitionsRevoked (Collection <TopicPartition> partitions) {
                  commitDBTransaction();
         }

         public void onPartitionsAssigned(Collection <TopicPartiton> partitions) {
              for(TopicPartition partition : partitions)
                  consumer.seek(partition, getOffsetFromDB(partition));
         }

     }
}   // these brackets are exactly the same as the book. I didn't change anything. You might want to though.    

   consumer.subscribe (topics, new SaveOffsetOnRebalance(consumer));
   consumer.poll(0);

   for ( TopicPartition partition : consumer.assignment())
       consumer.seek(partition, getOffsetFromDB(partition));

   while (true) {
         ConsumerRecords <String, String> records = consumer.poll(100);
         for (ConsumerRecord <String, String> record : records)
         { 
               processRecord(record);
               storeRecordInDB(record);
               storeOffsetInDB(record.topic(), record.partition(), record.offset());
         }
         commitDBTransaction();
   }

【讨论】：