kafkaoffset设置

from kafka import KafkaConsumer
from kafka import TopicPartition
from kafka.structs import OffsetAndMetadata
...
topic = 'your_topic'
partition = 0
tp = TopicPartition(topic,partition)
kafkaConsumer = KafkaConsumer(config here...)
kafkaConsumer.assign([tp])
offset = 15394125
kafkaConsumer.commit({
    tp: OffsetAndMetadata(offset, None)
})

meta = consumer.partitions_for_topic(topic)
options = {}
options[partition] = OffsetAndMetadata(message.offset, meta)
consumer.commit(options)

# a better way, remove assign partition manually, and extract partition info from kafka message

topic_partition = TopicPartition(command_params["topic"], message.partition) 
consumer.seek(topic_partition, offset_value) 
consumer.commit()