Skip to content
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

Jump to bottom

[ducktape] Fix connection rate test #6315

Merged
merged 2 commits into from
Sep 15, 2022
Merged

[ducktape] Fix connection rate test #6315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9e6b5b3
ducktape: Add retry_sec option to rpk_consumer
VadimPlh Sep 9, 2022
ede9131
ducktape: Use rpkconsumer in conn_rate test
VadimPlh Sep 9, 2022
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
11 changes: 6 additions & 5 deletions tests/rptest/services/rpk_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,8 @@ def __init__(self,
group='',
save_msgs=True,
fetch_max_bytes=None,
num_msgs=None):
num_msgs=None,
retry_sec=5):
super(RpkConsumer, self).__init__(context, num_nodes=1)
self._redpanda = redpanda
self._topic = topic
Expand All @@ -48,9 +49,9 @@ def __init__(self,
self._save_msgs = save_msgs
self._fetch_max_bytes = fetch_max_bytes
self._num_msgs = num_msgs
self._retry_sec = retry_sec

def _worker(self, idx, node):
retry_sec = 5
err = None

self._stopping.clear()
Expand All @@ -65,11 +66,11 @@ def _worker(self, idx, node):

err = e
self._redpanda.logger.error(
f"Consumer failed with error: '{e}'. Retrying in {retry_sec} seconds."
f"Consumer failed with error: '{e}'. Retrying in {self._retry_sec} seconds."
)
attempt += 1
self._stopping.wait(retry_sec)
time.sleep(retry_sec)
self._stopping.wait(self._retry_sec)
time.sleep(self._retry_sec)
else:
err = None

Expand Down
30 changes: 15 additions & 15 deletions tests/rptest/tests/connection_rate_limit_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from rptest.tests.prealloc_nodes import PreallocNodesTest
from rptest.services.redpanda import ResourceSettings
from rptest.services.kgo_verifier_services import KgoVerifierProducer
from rptest.services.kaf_consumer import KafConsumer
from rptest.services.rpk_consumer import RpkConsumer
from rptest.services.metrics_check import MetricCheck

RATE_METRIC = "vectorized_kafka_rpc_connections_wait_rate_total"
Expand Down Expand Up @@ -52,8 +52,12 @@ def __init__(self, test_context):
self.preallocated_nodes)

def start_consumer(self):
return KafConsumer(self.test_context, self.redpanda, self.topics[0],
self.READ_COUNT, "oldest")
return RpkConsumer(context=self.test_context,
redpanda=self.redpanda,
topic=self.topics[0],
num_msgs=1,
save_msgs=True,
retry_sec=(1 / self.RATE_LIMIT))

def stop_consumer(self, consumer):
try:
Expand All @@ -77,17 +81,15 @@ def consumed():
need_finish = True
for i in range(consumers_count):
self.logger.debug(
f"Offset for {i} consumer: {consumers[i].offset}")
f"Offset for {i} consumer: {len(consumers[i].messages)}")

if consumers[i].done is True:
self.stop_consumer(consumers[i])
self.logger.debug(f"Rerun consumer {i}")
consumers[i] = self.start_consumer()
consumers[i].start()

if (consumers[i].offset.get(0) is
None) or consumers[i].offset[0] == 0:
if len(consumers[i].messages) == 0:
need_finish = False
if consumers[i].done is True:
self.stop_consumer(consumers[i])
self.logger.debug(f"Rerun consumer {i}")
consumers[i] = self.start_consumer()
consumers[i].start()

return need_finish

Expand All @@ -105,7 +107,7 @@ def consumed():
def get_read_time(self, consumers_count):
deltas = list()

for i in range(10):
for i in range(20):
connection_time = self.read_data(consumers_count)
deltas.append(connection_time)

Expand All @@ -114,8 +116,6 @@ def get_read_time(self, consumers_count):

return sum(deltas) / len(deltas)

@ok_to_fail # https://github.com/redpanda-data/redpanda/issues/5276
# https://github.com/redpanda-data/redpanda/issues/6074
@cluster(num_nodes=8)
def connection_rate_test(self):
self._producer.start(clean=False)
Expand Down