Skip to content

Stabilize Kafka integration tests with explicit readiness checks #2736

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Copilot wants to merge 6 commits into
base: main
Choose a base branch
from
Draft

Stabilize Kafka integration tests with explicit readiness checks #2736

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b73a32a
Initial plan
Copilot Sep 11, 2026
6db7c52
Stabilize Kafka test readiness
Copilot Sep 11, 2026
e8a2523
Address Kafka test review feedback
Copilot Sep 11, 2026
ce1de2e
Finalize Kafka readiness helpers
Copilot Sep 11, 2026
f01418c
Bound Kafka consumer offset lookups
Copilot Sep 11, 2026
eb88b8a
Use fixed Kafka consume timeout
Copilot Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Use fixed Kafka consume timeout 
Co-authored-by: xrmx <12932+xrmx@users.noreply.github.com>
  • Loading branch information
@xrmx
Copilot and xrmx authored Sep 11, 2026
commit eb88b8a445bb8607f199889e5f385ed4ed321cc6
25 changes: 10 additions & 15 deletions tests/instrumentation/kafka_tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,20 +243,15 @@ def _produce_records(producer, records, elasticapm_client=None, transaction_type

def _consume_until_stop_iteration(consumer, on_message=None):
messages = []
original_consumer_timeout_ms = consumer.config["consumer_timeout_ms"]
consumer.config["consumer_timeout_ms"] = KAFKA_CONSUME_TIMEOUT_MS
try:
iterator = iter(consumer)
while True:
try:
message = next(iterator)
except StopIteration:
return messages
messages.append(message)
if on_message is not None:
on_message(message)
finally:
consumer.config["consumer_timeout_ms"] = original_consumer_timeout_ms
iterator = iter(consumer)
while True:
try:
message = next(iterator)
except StopIteration:
return messages
messages.append(message)
if on_message is not None:
on_message(message)


def _capture_span(name):
Expand Down Expand Up @@ -298,7 +293,7 @@ def producer(topics):
def consumer(topics):
consumer = KafkaConsumer(
bootstrap_servers=KAFKA_BOOTSTRAP_SERVER,
consumer_timeout_ms=500,
consumer_timeout_ms=KAFKA_CONSUME_TIMEOUT_MS,
# Unique topics can be produced to before the consumer's initial offset is resolved.
auto_offset_reset="earliest",
request_timeout_ms=KAFKA_OPERATION_TIMEOUT_MS,
Expand Down