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>

Author: Copilot

tests/instrumentation/kafka_tests.py

@@ -114,6 +114,7 @@ def delayed_send():
         producer.send("test", key=b"foo", value=b"bar")
         producer.send("test", key=b"baz", value=b"bazzinga")
         elasticapm_client.end_transaction("foo")
+        producer.flush()
 
     thread = threading.Thread(target=delayed_send)
     thread.start()
@@ -140,6 +141,7 @@ def delayed_send():
         producer.send("test", key=b"foo", value=b"bar")
         producer.send("test", key=b"baz", value=b"bazzinga")
         elasticapm_client.end_transaction("foo")
+        producer.flush()
 
     thread = threading.Thread(target=delayed_send)
     thread.start()
@@ -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()