Fix flaky kafka tests by flushing producer before ending sender transactions (#2619)

* Initial plan

* fix: add producer.flush() to kafka tests to prevent race conditions

The producer.send() method in kafka-python is asynchronous - it queues
messages to be sent by a background I/O thread. Without calling
producer.flush(), messages may not be delivered to the Kafka broker
before the consumer's consumer_timeout_ms=500 expires. This creates a
race condition where the consumer loop exits with no items consumed,
resulting in no spans being created.

Added producer.flush() after producer.send() calls in all delayed_send
functions to ensure messages are actually delivered to Kafka.

Agent-Logs-Url: https://github.com/elastic/apm-agent-python/sessions/e5ad59a6-a44a-4dd2-978e-fe4844b5c198

Co-authored-by: xrmx <12932+xrmx@users.noreply.github.com>

* fix: wait for kafka send futures in flaky consumer tests

Agent-Logs-Url: https://github.com/elastic/apm-agent-python/sessions/c8da0228-9a66-4391-a204-fe0b99b1b85b

Co-authored-by: xrmx <12932+xrmx@users.noreply.github.com>

* fix: flush kafka producer before ending sender transaction

Agent-Logs-Url: https://github.com/elastic/apm-agent-python/sessions/610642ed-3a8d-4371-bccb-485eaee1d69d

Co-authored-by: xrmx <12932+xrmx@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: xrmx <12932+xrmx@users.noreply.github.com>
Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: Copilot

tests/instrumentation/kafka_tests.py

@@ -113,6 +113,7 @@ def delayed_send():
         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)
@@ -139,6 +140,7 @@ def delayed_send():
         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)
@@ -166,6 +168,7 @@ def delayed_send():
         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()
@@ -186,6 +189,7 @@ def delayed_send():
         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()
@@ -205,6 +209,7 @@ def delayed_send():
         time.sleep(0.2)
         producer.send("test", key=b"foo", value=b"bar")
         producer.send("test", key=b"baz", value=b"bazzinga")
+        producer.flush()
 
     thread = threading.Thread(target=delayed_send)
     thread.start()
@@ -249,6 +254,7 @@ def test_kafka_consumer_unsampled_transaction_handles_stop_iteration(
     def delayed_send():
         time.sleep(0.2)
         producer.send("test", key=b"foo", value=b"bar")
+        producer.flush()
 
     thread = threading.Thread(target=delayed_send)
     thread.start()