Skip to content

Commit 0028893

Browse files
Make acknowledge APIs synchronous and improve the documents (#121)
Fixes #114 ### Motivation Currently the `acknowledge` and `acknowledge_cumulative` methods are all asynchronous. Even if any error happened, no exception would be raised. For example, when acknowledging cumulatively on a consumer whose consumer type is Shared or KeyShared, no error happens. ### Modifications - Change these methods to synchronous and raise exceptions if the acknowledgment failed. - Add `PulsarTest.test_acknowledge_failed` to test these failed cases. - Improve the documents to describe which exceptions could be raised in which cases.
1 parent 87a3506 commit 0028893

File tree

3 files changed

+41
-9
lines changed

3 files changed

+41
-9
lines changed

pulsar/__init__.py

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1305,6 +1305,11 @@ def acknowledge(self, message):
13051305
13061306
message:
13071307
The received message or message id.
1308+
1309+
Raises
1310+
------
1311+
OperationNotSupported
1312+
if `message` is not allowed to acknowledge
13081313
"""
13091314
if isinstance(message, Message):
13101315
self._consumer.acknowledge(message._message)
@@ -1324,6 +1329,11 @@ def acknowledge_cumulative(self, message):
13241329
13251330
message:
13261331
The received message or message id.
1332+
1333+
Raises
1334+
------
1335+
CumulativeAcknowledgementNotAllowedError
1336+
if the consumer type is ConsumerType.KeyShared or ConsumerType.Shared
13271337
"""
13281338
if isinstance(message, Message):
13291339
self._consumer.acknowledge_cumulative(message._message)

src/consumer.cc

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,12 @@ Messages Consumer_batch_receive(Consumer& consumer) {
5050
return msgs;
5151
}
5252

53-
void Consumer_acknowledge(Consumer& consumer, const Message& msg) { consumer.acknowledgeAsync(msg, nullptr); }
53+
void Consumer_acknowledge(Consumer& consumer, const Message& msg) {
54+
waitForAsyncResult([&](ResultCallback callback) { consumer.acknowledgeAsync(msg, callback); });
55+
}
5456

5557
void Consumer_acknowledge_message_id(Consumer& consumer, const MessageId& msgId) {
56-
Py_BEGIN_ALLOW_THREADS consumer.acknowledgeAsync(msgId, nullptr);
57-
Py_END_ALLOW_THREADS
58+
waitForAsyncResult([&](ResultCallback callback) { consumer.acknowledgeAsync(msgId, callback); });
5859
}
5960

6061
void Consumer_negative_acknowledge(Consumer& consumer, const Message& msg) {
@@ -63,18 +64,16 @@ void Consumer_negative_acknowledge(Consumer& consumer, const Message& msg) {
6364
}
6465

6566
void Consumer_negative_acknowledge_message_id(Consumer& consumer, const MessageId& msgId) {
66-
Py_BEGIN_ALLOW_THREADS consumer.negativeAcknowledge(msgId);
67-
Py_END_ALLOW_THREADS
67+
waitForAsyncResult([&](ResultCallback callback) { consumer.acknowledgeAsync(msgId, callback); });
6868
}
6969

7070
void Consumer_acknowledge_cumulative(Consumer& consumer, const Message& msg) {
71-
Py_BEGIN_ALLOW_THREADS consumer.acknowledgeCumulativeAsync(msg, nullptr);
72-
Py_END_ALLOW_THREADS
71+
waitForAsyncResult([&](ResultCallback callback) { consumer.acknowledgeCumulativeAsync(msg, callback); });
7372
}
7473

7574
void Consumer_acknowledge_cumulative_message_id(Consumer& consumer, const MessageId& msgId) {
76-
Py_BEGIN_ALLOW_THREADS consumer.acknowledgeCumulativeAsync(msgId, nullptr);
77-
Py_END_ALLOW_THREADS
75+
waitForAsyncResult(
76+
[&](ResultCallback callback) { consumer.acknowledgeCumulativeAsync(msgId, callback); });
7877
}
7978

8079
void Consumer_close(Consumer& consumer) {

tests/pulsar_test.py

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1437,6 +1437,29 @@ def send_callback(res, msg):
14371437
producer.flush()
14381438
client.close()
14391439

1440+
def test_acknowledge_failed(self):
1441+
client = Client(self.serviceUrl)
1442+
topic = 'test_acknowledge_failed'
1443+
producer = client.create_producer(topic)
1444+
consumer1 = client.subscribe(topic, 'sub1', consumer_type=ConsumerType.Shared)
1445+
consumer2 = client.subscribe(topic, 'sub2', consumer_type=ConsumerType.KeyShared)
1446+
msg_id = producer.send('hello'.encode())
1447+
msg1 = consumer1.receive()
1448+
with self.assertRaises(pulsar.CumulativeAcknowledgementNotAllowedError):
1449+
consumer1.acknowledge_cumulative(msg1)
1450+
with self.assertRaises(pulsar.CumulativeAcknowledgementNotAllowedError):
1451+
consumer1.acknowledge_cumulative(msg1.message_id())
1452+
msg2 = consumer2.receive()
1453+
with self.assertRaises(pulsar.CumulativeAcknowledgementNotAllowedError):
1454+
consumer2.acknowledge_cumulative(msg2)
1455+
with self.assertRaises(pulsar.CumulativeAcknowledgementNotAllowedError):
1456+
consumer2.acknowledge_cumulative(msg2.message_id())
1457+
consumer = client.subscribe([topic, topic + '-another'], 'sub')
1458+
# The message id does not have a topic name
1459+
with self.assertRaises(pulsar.OperationNotSupported):
1460+
consumer.acknowledge(msg_id)
1461+
client.close()
1462+
14401463

14411464
if __name__ == "__main__":
14421465
main()

0 commit comments

Comments
 (0)