Flaky kafka-newest test on 3.14

[xrmx]

____________________ test_kafka_consume_ongoing_transaction ____________________

instrument = None
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f408d41d9d0>
producer = <kafka.producer.kafka.KafkaProducer object at 0x7f408d41ecf0>
consumer = <kafka.consumer.group.KafkaConsumer object at 0x7f408c2a1ba0>
topics = ['test', 'foo', 'bar']

    def test_kafka_consume_ongoing_transaction(instrument, elasticapm_client, producer, consumer, topics):
        def delayed_send():
            time.sleep(0.2)
            elasticapm_client.begin_transaction("foo")
            producer.send("test", key=b"foo", value=b"bar")
            producer.send("test", key=b"baz", value=b"bazzinga")
            producer.flush()
            elasticapm_client.end_transaction("foo")
    
        thread = threading.Thread(target=delayed_send)
        thread.start()
        transaction = elasticapm_client.begin_transaction("foo")
        for item in consumer:
            pass
        thread.join()
        elasticapm_client.end_transaction("foo")
        transactions = elasticapm_client.events[TRANSACTION]
        assert len(transactions) == 2
        external_spans = elasticapm_client.spans_for_transaction(transactions[0])
        spans = elasticapm_client.spans_for_transaction(transactions[1])
        assert len(external_spans) == 2
>       assert len(spans) == 2
E       assert 0 == 2
E        +  where 0 = len([])

consumer   = <kafka.consumer.group.KafkaConsumer object at 0x7f408c2a1ba0>
delayed_send = <function test_kafka_consume_ongoing_transaction.<locals>.delayed_send at 0x7f408c16c1a0>
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f408d41d9d0>
external_spans = [{'action': 'send', 'context': {'destination': {'address': 'kafka', 'port': 9092, 'service': {'name': 'kafka', 'resour...vice': {'target': {'name': 'test', 'type': 'kafka'}}}, 'duration': 0.48899999999999993, 'id': 'f964a73fb9260ef4', ...}]
instrument = None
producer   = <kafka.producer.kafka.KafkaProducer object at 0x7f408d41ecf0>
spans      = []
thread     = <Thread(Thread-12 (delayed_send), stopped 139915119552192)>
topics     = ['test', 'foo', 'bar']
transaction = <elasticapm.traces.Transaction object at 0x7f408d3ffd90>
transactions = [{'context': {'tags': {}}, 'duration': 12.602, 'id': '0a2d2ff318df135c', 'name': 'foo', ...}, {'context': {'tags': {}}, 'duration': 502.87899999999996, 'id': '60f261ba230a8914', 'name': 'foo', ...}]

[xrmx]

Also this for 3.13 and kafka-newest:

_______________________ test_kafka_consumer_ignore_topic _______________________

instrument = None
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f8974c3ecf0>
producer = <kafka.producer.kafka.KafkaProducer object at 0x7f8974c3fbd0>
consumer = <kafka.consumer.group.KafkaConsumer object at 0x7f8974c2e7b0>
topics = ['test', 'foo', 'bar']

    def test_kafka_consumer_ignore_topic(instrument, elasticapm_client, producer, consumer, topics):
        elasticapm_client.config.update("1", ignore_message_queues="foo*,*bar")
    
        def delayed_send():
            time.sleep(0.2)
            producer.send(topic="foo", key=b"foo", value=b"bar")
            producer.send("bar", key=b"foo", value=b"bar")
            producer.send("test", key=b"foo", value=b"bar")
            producer.flush()
    
        thread = threading.Thread(target=delayed_send)
        thread.start()
        for item in consumer:
            with elasticapm.capture_span("test"):
                assert item
        thread.join()
        transactions = elasticapm_client.events[TRANSACTION]
>       assert len(transactions) == 1
E       assert 0 == 1
E        +  where 0 = len([])

consumer   = <kafka.consumer.group.KafkaConsumer object at 0x7f8974c2e7b0>
delayed_send = <function test_kafka_consumer_ignore_topic.<locals>.delayed_send at 0x7f897170d260>
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f8974c3ecf0>
instrument = None
item       = ConsumerRecord(topic='bar', partition=0, leader_epoch=0, offset=0, timestamp=1777550213835, timestamp_type=0, key=b'foo', value=b'bar', headers=[], checksum=None, serialized_key_size=3, serialized_value_size=3, serialized_header_size=-1)
producer   = <kafka.producer.kafka.KafkaProducer object at 0x7f8974c3fbd0>
thread     = <Thread(Thread-11 (delayed_send), stopped 140228290295488)>
topics     = ['test', 'foo', 'bar']
transactions = []