kafka_prefix = 'siaiajkvox.swh.journal.objects' kafka_server = '127.0.0.1:55629' consumer = <cimpl.Consumer object at 0x7fc4456ae620> large_directories = [Directory(entries=[DirectoryEntry(name=b'000000000', type='file', target=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x...x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', perms=33188)], id=b').\x887-c?\x10\x9eRa\x93\t\x16\xeb_\x02\xbaRz'), ...] def test_write_large_objects( kafka_prefix: str, kafka_server: str, consumer: Consumer, large_directories: List[Directory], ): kafka_prefix += ".swh.journal.objects" # Needed as there is no directories in TEST_OBJECT_DICTS, the consumer # isn't autosubscribed to directories. consumer.subscribe([kafka_prefix + ".directory"]) writer = KafkaJournalWriter( brokers=[kafka_server], client_id="kafka_writer", prefix=kafka_prefix, ) writer.write_additions("directory", large_directories) > consumed_messages = consume_messages(consumer, kafka_prefix, len(large_directories)) .tox/py3/lib/python3.7/site-packages/swh/journal/tests/test_kafka_writer.py:182: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ consumer = <cimpl.Consumer object at 0x7fc4456ae620> kafka_prefix = 'siaiajkvox.swh.journal.objects', expected_messages = 21 def consume_messages(consumer, kafka_prefix, expected_messages): """Consume expected_messages from the consumer; Sort them all into a consumed_objects dict""" consumed_messages = defaultdict(list) fetched_messages = 0 retries_left = 1000 while fetched_messages < expected_messages: if retries_left == 0: > raise ValueError("Timed out fetching messages from kafka") E ValueError: Timed out fetching messages from kafka .tox/py3/lib/python3.7/site-packages/swh/journal/tests/test_kafka_writer.py:33: ValueError