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 = []

[xrmx]

______________________________ test_kafka_consume ______________________________      
                                                                                                                                           
    instrument = None                                                                                                                      
    elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f2655d63820>                                                          
    producer = <kafka.producer.kafka.KafkaProducer object at 0x7f2655898770>                                                               
    consumer = <kafka.consumer.group.KafkaConsumer object at 0x7f2655898e90>                                                               
    topics = ['test-5dd55f08e81a44eeb0fb8c78723ad86e', 'foo-5dd55f08e81a44eeb0fb8c78723ad86e', '5dd55f08e81a44eeb0fb8c78723ad86e-bar']     
                                                                                                                                           
        def test_kafka_consume(instrument, elasticapm_client, producer, consumer, topics):                                                 
            test_topic = topics[0]                                                                                                         
                                                                                                                                           
            def delayed_send():                                                                                                            
                time.sleep(0.2)                                                                                                            
                elasticapm_client.begin_transaction("foo")                                                                                 
                producer.send(test_topic, key=b"foo", value=b"bar")                                                                        
                producer.send(test_topic, key=b"baz", value=b"bazzinga")                                                                   
                producer.flush()                                                                                                           
                elasticapm_client.end_transaction("foo")                                                                                   
                                                                                                                                           
            thread = threading.Thread(target=delayed_send)                                                                                 
            thread.start()                                                                                                                 
            for item in consumer:                                                                                                          
                with elasticapm.capture_span("foo"):                                                                                       
                    pass                                                                                                                   
            thread.join()                                                                                                                  
            transactions = elasticapm_client.events[TRANSACTION]                                                                           
            spans = elasticapm_client.events[SPAN]                                                                                         
            # the consumer transactions should have the same trace id as the transaction that triggered the messages                       
            assert transactions[0]["trace_id"] == transactions[1]["trace_id"] == transactions[2]["trace_id"]                               
            assert transactions[1]["name"] == f"Kafka RECEIVE from {test_topic}"                                                           
            assert transactions[1]["type"] == "messaging"                                                                                  
            assert transactions[1]["context"]["message"]["queue"]["name"] == test_topic                                                    
                                                                                                                                           
    >       assert spans[2]["transaction_id"] == transactions[1]["id"]                                                                     
    E       AssertionError: assert '60cc8d6b269810f1' == 'ed307accab7bd841'                                                                
    E         - ed307accab7bd841                                                                                                           
    E         + 60cc8d6b269810f1

[xrmx]

___________________ ERROR at teardown of test_kafka_produce ____________________

topics = ['test-bd4dab29c1d546249d2c415a81db2787', 'foo-bd4dab29c1d546249d2c415a81db2787', 'bd4dab29c1d546249d2c415a81db2787-bar']

    @pytest.fixture()
    def producer(topics):
        producer = KafkaProducer(bootstrap_servers=f"{KAFKA_HOST}:9092")
        yield producer
>       producer.close()

producer   = <kafka.producer.kafka.KafkaProducer object at 0x7f0c3194b350>
topics     = ['test-bd4dab29c1d546249d2c415a81db2787', 'foo-bd4dab29c1d546249d2c415a81db2787', 'bd4dab29c1d546249d2c415a81db2787-bar']

tests/instrumentation/kafka_tests.py:75: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/home/user/.local/lib/python3.7/site-packages/kafka/producer/kafka.py:646: in close
    self._sender.initiate_close()
        invoked_from_callback = False
        null_logger = False
        self       = <kafka.producer.kafka.KafkaProducer object at 0x7f0c3194b350>
        timeout    = 9223372036.0
/home/user/.local/lib/python3.7/site-packages/kafka/producer/sender.py:349: in initiate_close
    self.wakeup()
        self       = <Sender(kafka-python-producer-1-network-thread, stopped daemon 139690344281792)>
/home/user/.local/lib/python3.7/site-packages/kafka/producer/sender.py:631: in wakeup
    self._client.wakeup()
        self       = <Sender(kafka-python-producer-1-network-thread, stopped daemon 139690344281792)>
/home/user/.local/lib/python3.7/site-packages/kafka/client_async.py:1127: in wakeup
    raise e
        self       = <kafka.client_async.KafkaClient object at 0x7f0c31b55450>

[xrmx]

______________________________ test_kafka_consume ______________________________

instrument = None
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7fdd6b7aedc0>
producer = <kafka.producer.kafka.KafkaProducer object at 0x7fdd6ba9fd30>
consumer = <kafka.consumer.group.KafkaConsumer object at 0x7fdd6bace3d0>
topics = ['test-85fe8b7d41a54316bcb5396b18d13eb2', 'foo-85fe8b7d41a54316bcb5396b18d13eb2', '85fe8b7d41a54316bcb5396b18d13eb2-bar']

    def test_kafka_consume(instrument, elasticapm_client, producer, consumer, topics):
        test_topic = topics[0]
    
        def delayed_send():
            time.sleep(0.2)
            elasticapm_client.begin_transaction("foo")
            producer.send(test_topic, key=b"foo", value=b"bar")
            producer.send(test_topic, key=b"baz", value=b"bazzinga")
            producer.flush()
            elasticapm_client.end_transaction("foo")
    
        thread = threading.Thread(target=delayed_send)
        thread.start()
        for item in consumer:
            with elasticapm.capture_span("foo"):
                pass
        thread.join()
        transactions = elasticapm_client.events[TRANSACTION]
        spans = elasticapm_client.events[SPAN]
        # the consumer transactions should have the same trace id as the transaction that triggered the messages
>       assert transactions[0]["trace_id"] == transactions[1]["trace_id"] == transactions[2]["trace_id"]
E       IndexError: list index out of range

consumer   = <kafka.consumer.group.KafkaConsumer object at 0x7fdd6bace3d0>
delayed_send = <function test_kafka_consume.<locals>.delayed_send at 0x7fdd6bac3e50>
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7fdd6b7aedc0>
instrument = None
producer   = <kafka.producer.kafka.KafkaProducer object at 0x7fdd6ba9fd30>
spans      = [{'action': 'send', 'context': {'destination': {'address': 'kafka', 'port': 9092, 'service': {'name': 'kafka', 'resour...85fe8b7d41a54316bcb5396b18d13eb2', 'type': 'kafka'}}}, 'duration': 0.16799999999999998, 'id': 'a60ca35604f6c14c', ...}]
test_topic = 'test-85fe8b7d41a54316bcb5396b18d13eb2'
thread     = <Thread(Thread-11, stopped 140588949513920)>
topics     = ['test-85fe8b7d41a54316bcb5396b18d13eb2', 'foo-85fe8b7d41a54316bcb5396b18d13eb2', '85fe8b7d41a54316bcb5396b18d13eb2-bar']
transactions = [{'context': {'tags': {}}, 'duration': 7.6819999999999995, 'id': '708e3e88826c04cb', 'name': 'foo', ...}]

[xrmx]

This should be hopefully improved with #2657 , let's see if that was enough.

[xrmx]

_____________ test_kafka_consumer_ignore_topic_ongoing_transaction _____________

instrument = None
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f00e2386610>
producer = <kafka.producer.kafka.KafkaProducer object at 0x7f00e1e144c0>
consumer = <kafka.consumer.group.KafkaConsumer object at 0x7f00e1e2c580>
topics = ['test-fee634359f6f4757bd8cbba69688c5f9', 'foo-fee634359f6f4757bd8cbba69688c5f9', 'fee634359f6f4757bd8cbba69688c5f9-bar']

    def test_kafka_consumer_ignore_topic_ongoing_transaction(instrument, elasticapm_client, producer, consumer, topics):
        test_topic, foo_topic, bar_topic = topics
        elasticapm_client.config.update("1", ignore_message_queues="foo*,*bar")
    
        def delayed_send():
            time.sleep(0.2)
            producer.send(topic=foo_topic, key=b"foo", value=b"bar")
            producer.send(bar_topic, key=b"foo", value=b"bar")
            producer.send(test_topic, key=b"foo", value=b"bar")
            producer.flush()
    
        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]
        spans = elasticapm_client.spans_for_transaction(transactions[0])
>       assert len(spans) == 1
E       assert 0 == 1
E        +  where 0 = len([])

bar_topic  = 'fee634359f6f4757bd8cbba69688c5f9-bar'
consumer   = <kafka.consumer.group.KafkaConsumer object at 0x7f00e1e2c580>
delayed_send = <function test_kafka_consumer_ignore_topic_ongoing_transaction.<locals>.delayed_send at 0x7f00e275c280>
elasticapm_client = <tests.fixtures.TempStoreClient object at 0x7f00e2386610>
foo_topic  = 'foo-fee634359f6f4757bd8cbba69688c5f9'
instrument = None
item       = ConsumerRecord(topic='fee634359f6f4757bd8cbba69688c5f9-bar', partition=0, leader_epoch=0, offset=0, timestamp=17797243...o', 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 0x7f00e1e144c0>
spans      = []
test_topic = 'test-fee634359f6f4757bd8cbba69688c5f9'
thread     = <Thread(Thread-2, stopped 139641667712704)>
topics     = ['test-fee634359f6f4757bd8cbba69688c5f9', 'foo-fee634359f6f4757bd8cbba69688c5f9', 'fee634359f6f4757bd8cbba69688c5f9-bar']
transaction = <elasticapm.traces.Transaction object at 0x7f00e23d48b0>
transactions = [{'context': {'tags': {}}, 'duration': 817.978, 'id': '0f43fcdaadea211e', 'name': 'foo', ...}]

tests/instrumentation/kafka_tests.py:233: AssertionError