self = <swh.search.tests.test_cli.CliTestCase testMethod=test__journal_client__origin> def test__journal_client__origin(self): """Tests the re-indexing when origin_batch_size*task_batch_size is a divisor of nb_origins.""" topic = "swh.journal.objects.origin" value = value_to_kafka({"url": "http://foobar.baz",}) message = MagicMock() message.error.return_value = None message.topic.return_value = topic message.value.return_value = value mock_consumer = MockedKafkaConsumerWithTopics([message]) with patch("swh.journal.client.Consumer", return_value=mock_consumer): result = invoke( False, ["journal-client", "objects", "--stop-after-objects", "1",], JOURNAL_OBJECTS_CONFIG, > elasticsearch_host=self._elasticsearch_host, ) .tox/py3/lib/python3.7/site-packages/swh/search/tests/test_cli.py:79: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .tox/py3/lib/python3.7/site-packages/swh/search/tests/test_cli.py:57: in invoke raise result.exception .tox/py3/lib/python3.7/site-packages/click/testing.py:329: in invoke cli.main(args=args or (), prog_name=prog_name, **extra) .tox/py3/lib/python3.7/site-packages/click/core.py:782: in main rv = self.invoke(ctx) .tox/py3/lib/python3.7/site-packages/click/core.py:1259: in invoke return _process_result(sub_ctx.command.invoke(sub_ctx)) .tox/py3/lib/python3.7/site-packages/click/core.py:1259: in invoke return _process_result(sub_ctx.command.invoke(sub_ctx)) .tox/py3/lib/python3.7/site-packages/click/core.py:1066: in invoke return ctx.invoke(self.callback, **ctx.params) .tox/py3/lib/python3.7/site-packages/click/core.py:610: in invoke return callback(*args, **kwargs) .tox/py3/lib/python3.7/site-packages/click/decorators.py:21: in new_func return f(get_current_context(), *args, **kwargs) .tox/py3/lib/python3.7/site-packages/swh/search/cli.py:72: in journal_client_objects **journal_cfg, .tox/py3/lib/python3.7/site-packages/swh/journal/client.py:37: in get_journal_client return JournalClient(**kwargs) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <swh.journal.client.JournalClient object at 0x7f2fa376ad30> brokers = ['192.0.2.1'], group_id = 'test-consumer' prefix = 'swh.journal.objects', object_types = ['origin', 'origin_visit'] stop_after_objects = 1, batch_size = 200, process_timeout = None auto_offset_reset = 'earliest', stop_on_eof = False, kwargs = {} debug_logging = False, group_instance_id = None consumer_settings = {'auto.offset.reset': 'earliest', 'bootstrap.servers': '192.0.2.1', 'enable.auto.commit': False, 'error_cb': <function _error_cb at 0x7f2fa3b57e18>, ...} def __init__( self, brokers: Union[str, List[str]], group_id: str, prefix: Optional[str] = None, object_types: Optional[List[str]] = None, stop_after_objects: Optional[int] = None, batch_size: int = 200, process_timeout: Optional[float] = None, auto_offset_reset: str = "earliest", stop_on_eof: bool = False, **kwargs, ): if prefix is None: prefix = DEFAULT_PREFIX if auto_offset_reset not in ACCEPTED_OFFSET_RESET: raise ValueError( "Option 'auto_offset_reset' only accept %s, not %s" % (ACCEPTED_OFFSET_RESET, auto_offset_reset) ) if batch_size <= 0: raise ValueError("Option 'batch_size' needs to be positive") self.value_deserializer = kafka_to_value if isinstance(brokers, str): brokers = [brokers] debug_logging = rdkafka_logger.isEnabledFor(logging.DEBUG) if debug_logging and "debug" not in kwargs: kwargs["debug"] = "consumer" # Static group instance id management group_instance_id = os.environ.get("KAFKA_GROUP_INSTANCE_ID") if group_instance_id: kwargs["group.instance.id"] = group_instance_id if "group.instance.id" in kwargs: # When doing static consumer group membership, set a higher default # session timeout. The session timeout is the duration after which # the broker considers that a consumer has left the consumer group # for good, and triggers a rebalance. Considering our current # processing pattern, 10 minutes gives the consumer ample time to # restart before that happens. if "session.timeout.ms" not in kwargs: kwargs["session.timeout.ms"] = 10 * 60 * 1000 # 10 minutes if "session.timeout.ms" in kwargs: # When the session timeout is set, rdkafka requires the max poll # interval to be set to a higher value; the max poll interval is # rdkafka's way of figuring out whether the client's message # processing thread has stalled: when the max poll interval lapses # between two calls to consumer.poll(), rdkafka leaves the consumer # group and terminates the connection to the brokers. # # We default to 1.5 times the session timeout if "max.poll.interval.ms" not in kwargs: kwargs["max.poll.interval.ms"] = kwargs["session.timeout.ms"] // 2 * 3 consumer_settings = { **kwargs, "bootstrap.servers": ",".join(brokers), "auto.offset.reset": auto_offset_reset, "group.id": group_id, "on_commit": _on_commit, "error_cb": _error_cb, "enable.auto.commit": False, "logger": rdkafka_logger, } self.stop_on_eof = stop_on_eof if self.stop_on_eof: consumer_settings["enable.partition.eof"] = True logger.debug("Consumer settings: %s", consumer_settings) self.consumer = Consumer(consumer_settings) > existing_topics = self.consumer.list_topics(timeout=10).topics.keys() E AttributeError: 'set' object has no attribute 'topics' .tox/py3/lib/python3.7/site-packages/swh/journal/client.py:168: AttributeError