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

Error if connections_max_idle_ms not larger than request_timeout_ms #1688

Merged
merged 1 commit into from Mar 15, 2019
Merged

Error if connections_max_idle_ms not larger than request_timeout_ms #1688

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
10 changes: 7 additions & 3 deletions kafka/consumer/group.py
View file
Open in desktop
Expand Up @@ -313,11 +313,15 @@ def __init__(self, *topics, **configs):
new_config, self.config['auto_offset_reset'])
self.config['auto_offset_reset'] = new_config

connections_max_idle_ms = self.config['connections_max_idle_ms']
request_timeout_ms = self.config['request_timeout_ms']
fetch_max_wait_ms = self.config['fetch_max_wait_ms']
if request_timeout_ms <= fetch_max_wait_ms:
raise KafkaConfigurationError("Request timeout (%s) must be larger than fetch-max-wait-ms (%s)" %
(request_timeout_ms, fetch_max_wait_ms))
if not (fetch_max_wait_ms < request_timeout_ms < connections_max_idle_ms):
raise KafkaConfigurationError(
"connections_max_idle_ms ({}) must be larger than "
"request_timeout_ms ({}) which must be larger than "
"fetch_max_wait_ms ({})."
.format(connections_max_idle_ms, request_timeout_ms, fetch_max_wait_ms))

metrics_tags = {'client-id': self.config['client_id']}
metric_config = MetricConfig(samples=self.config['metrics_num_samples'],
Expand Down
8 changes: 6 additions & 2 deletions test/test_consumer.py
View file
Open in desktop
Expand Up @@ -15,11 +15,15 @@
class TestKafkaConsumer:
def test_session_timeout_larger_than_request_timeout_raises(self):
with pytest.raises(KafkaConfigurationError):
KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0,9), group_id='foo', session_timeout_ms=60000, request_timeout_ms=40000)
KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), group_id='foo', session_timeout_ms=50000, request_timeout_ms=40000)

def test_fetch_max_wait_larger_than_request_timeout_raises(self):
with pytest.raises(KafkaConfigurationError):
KafkaConsumer(bootstrap_servers='localhost:9092', fetch_max_wait_ms=41000, request_timeout_ms=40000)
KafkaConsumer(bootstrap_servers='localhost:9092', fetch_max_wait_ms=50000, request_timeout_ms=40000)
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this was a simple change from 41000 to 50000 purely for improved readability/consistency with the other tests here, the test passes either way.


def test_request_timeout_larger_than_connections_max_idle_ms_raises(self):
with pytest.raises(KafkaConfigurationError):
KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), request_timeout_ms=50000, connections_max_idle_ms=40000)
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do not understand why this test fails with NoBrokerFound if api_version=(0, 9) is omitted... or rather, I realize it's a problem with the version probing timing out, but why does test_fetch_max_wait_larger_than_request_timeout_raises() right above succeed without needing to specify the version?

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that may have been a side effect of the old AND check. If you drop the api_version here I would expect the test to still pass

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIRC, I actually did try that exact thing this morning, and it still failed with the NoBrokerFound error. That's what confuses me.


def test_subscription_copy(self):
consumer = KafkaConsumer('foo', api_version=(0, 10))
Expand Down