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>

Author: Copilot

tests/instrumentation/kafka_tests.py

@@ -111,8 +111,9 @@ def test_kafka_consume(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").get(timeout=5)
-        producer.send("test", key=b"baz", value=b"bazzinga").get(timeout=5)
+        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)
@@ -137,8 +138,9 @@ def test_kafka_consume_ongoing_transaction(instrument, elasticapm_client, produc
     def delayed_send():
         time.sleep(0.2)
         elasticapm_client.begin_transaction("foo")
-        producer.send("test", key=b"foo", value=b"bar").get(timeout=5)
-        producer.send("test", key=b"baz", value=b"bazzinga").get(timeout=5)
+        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)
@@ -163,9 +165,10 @@ def test_kafka_consumer_ignore_topic(instrument, elasticapm_client, producer, co
 
     def delayed_send():
         time.sleep(0.2)
-        producer.send(topic="foo", key=b"foo", value=b"bar").get(timeout=5)
-        producer.send("bar", key=b"foo", value=b"bar").get(timeout=5)
-        producer.send("test", key=b"foo", value=b"bar").get(timeout=5)
+        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()
@@ -183,9 +186,10 @@ def test_kafka_consumer_ignore_topic_ongoing_transaction(instrument, elasticapm_
 
     def delayed_send():
         time.sleep(0.2)
-        producer.send(topic="foo", key=b"foo", value=b"bar").get(timeout=5)
-        producer.send("bar", key=b"foo", value=b"bar").get(timeout=5)
-        producer.send("test", key=b"foo", value=b"bar").get(timeout=5)
+        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()
@@ -203,8 +207,9 @@ def delayed_send():
 def test_kafka_poll_ongoing_transaction(instrument, elasticapm_client, producer, consumer, topics):
     def delayed_send():
         time.sleep(0.2)
-        producer.send("test", key=b"foo", value=b"bar").get(timeout=5)
-        producer.send("test", key=b"baz", value=b"bazzinga").get(timeout=5)
+        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()
@@ -248,7 +253,8 @@ 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").get(timeout=5)
+        producer.send("test", key=b"foo", value=b"bar")
+        producer.flush()
 
     thread = threading.Thread(target=delayed_send)
     thread.start()