Читай конкретную запись с пикафкой - PullRequest
Новая
       24

Читай конкретную запись с пикафкой

0 голосов
/ 16 мая 2018

Я хочу хранить большие файлы в Кафке, используя метаданные о записи, чтобы извлечь их в будущем.

Поэтому я рассылаю сообщения, содержащие тему, partition_id, offset, а затем пытаюсь получить файлтаким образом: 

def retrieve_file_from_kafka(topic_name, partition_id, offset):
    client = KafkaClient(hosts=BROKER_ADDRESS, broker_version="0.10.1.0")
    topic = client.topics[bytes(topic_name, "UTF-8")]

    consumer = topic.get_balanced_consumer(
    consumer_group=bytes("file_retrieve" + topic_name + str(partition_id) + str(offset), "UTF-8"))
    consumer.reset_offsets([(topic.partitions[partition_id], offset)])
    return consumer.consume()

Это не работает, хотя и просто печатает: 

Offset reset for partition 0 to timestamp 8 failed. Setting partition 0's internal counter to 8

Эта ошибка довольно загадочная, и это происходит на reset_offsets.Когда я пытаюсь потреблять, процесс застревает в ожидании rebalancing_lock.Что я делаю не так?

...