BewareMyPower commented on code in PR #109:
URL:
https://github.com/apache/pulsar-client-python/pull/109#discussion_r1207945154
##########
tests/pulsar_test.py:
##########
@@ -1437,6 +1439,135 @@ def send_callback(res, msg):
producer.flush()
client.close()
+ def test_keyshare_policy(self):
+ with self.assertRaises(ValueError):
+ # Raise error because sticky ranges are not provided.
+ pulsar.ConsumerKeySharedPolicy(
+ key_shared_mode=pulsar.KeySharedMode.Sticky,
+ allow_out_of_order_delivery=False,
+ )
+
+ expected_key_shared_mode = pulsar.KeySharedMode.Sticky
+ expected_allow_out_of_order_delivery = True
+ expected_sticky_ranges = [(0, 100), (101,200)]
+ consumer_key_shared_policy = pulsar.ConsumerKeySharedPolicy(
+ key_shared_mode=expected_key_shared_mode,
+ allow_out_of_order_delivery=expected_allow_out_of_order_delivery,
+ sticky_ranges=expected_sticky_ranges
+ )
+
+ self.assertEqual(consumer_key_shared_policy.key_shared_mode,
expected_key_shared_mode)
+
self.assertEqual(consumer_key_shared_policy.allow_out_of_order_delivery,
expected_allow_out_of_order_delivery)
+ self.assertEqual(consumer_key_shared_policy.sticky_ranges,
expected_sticky_ranges)
+
+ def test_keyshared_invalid_sticky_ranges(self):
+ client = Client(self.serviceUrl)
+ topic = "my-python-topic-keyshare-invalid-" + str(time.time())
+ with self.assertRaises(ValueError):
+ consumer_key_shared_policy = pulsar.ConsumerKeySharedPolicy(
+ key_shared_mode=pulsar.KeySharedMode.Sticky,
+ allow_out_of_order_delivery=False,
+ sticky_ranges=[(0,65536)]
+ )
+ client.subscribe(topic, "my-sub",
consumer_type=ConsumerType.KeyShared,
+ start_message_id_inclusive=True,
+ key_shared_policy=consumer_key_shared_policy)
+
+ with self.assertRaises(ValueError):
+ consumer_key_shared_policy = pulsar.ConsumerKeySharedPolicy(
+ key_shared_mode=pulsar.KeySharedMode.Sticky,
+ allow_out_of_order_delivery=False,
+ sticky_ranges=[(0, 100), (50, 150)]
+ )
+ client.subscribe(topic, "my-sub",
consumer_type=ConsumerType.KeyShared,
+ start_message_id_inclusive=True,
+ key_shared_policy=consumer_key_shared_policy)
+
+ def test_keyshared_autosplit(self):
+ client = Client(self.serviceUrl)
+ topic = "my-python-topic-keyshare-autosplit-" + str(time.time())
+ consumer_key_shared_policy = pulsar.ConsumerKeySharedPolicy(
+ key_shared_mode=pulsar.KeySharedMode.AutoSplit,
+ allow_out_of_order_delivery=True,
+ )
+
Review Comment:
```suggestion
```
Remove the line that is consist of all blank characters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
