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

Add "num_replicas" field for consumer #326

Closed
wants to merge 3 commits into from
Closed

Add "num_replicas" field for consumer #326

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3d98928
Add num_replicas field for consumer
Jul 6, 2022
de77557
Add consumer name
Jul 6, 2022
f0bbb69
Fix subject name when publishing
Jul 6, 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
1 change: 1 addition & 0 deletions nats/js/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,6 +365,7 @@ class ConsumerConfig(Base):
flow_control: Optional[bool] = None
idle_heartbeat: Optional[float] = None
headers_only: Optional[bool] = None
num_replicas: Optional[int] = None

@classmethod
def from_response(cls, resp: Dict[str, Any]):
Expand Down
19 changes: 19 additions & 0 deletions tests/test_js.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1577,3 +1577,22 @@ async def error_handler(e):

with pytest.raises(BadBucketError):
await js.key_value(bucket="TEST3")

class ConsumerReplicasTest(SingleJetStreamServerTestCase):
@async_test
async def test_number_of_consumer_replicas(self):
nc = await nats.connect()

js = nc.jetstream()
await js.add_stream(name="TESTREPLICAS", subjects=["test.replicas"])
Copy link

@shiv4289 shiv4289 Jul 7, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Hi @wallyqs Is there a way to make this stream R3 in the test case here?
This would strengthen the test so the streams is R3 and consumer is R1 and we can be 100% sure the replication factor for consumer is not inherited from the stream.

Copy link
Member

Choose a reason for hiding this comment

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

I'd like that too, would have to make the tests run in cluster mode and right now they only run in single node (no helper for that in the tests atm)

for i in range(0, 10):
await js.publish("test", f'{i}'.encode())
wallyqs marked this conversation as resolved.
Show resolved Hide resolved


# Create consumer
config = nats.js.api.ConsumerConfig(num_replicas=1, durable_name="mycons")
cons = await js.add_consumer(stream="TESTREPLICAS", config=config)

assert cons.config.num_replicas == 1

await nc.close()