This is an automated email from the ASF dual-hosted git repository.

penghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 3f36544  [Issue 3127][python-client] Replace Exceptions with 
PulsarExceptions (#7600)
3f36544 is described below

commit 3f36544c76620a7c6cd91c8047ecc28dbc3f85db
Author: Livio BenĨik <lbenc...@gmail.com>
AuthorDate: Sat May 15 04:15:48 2021 +0200

    [Issue 3127][python-client] Replace Exceptions with PulsarExceptions (#7600)
    
    Fixes #7600
    
    ### Motivation
    
    As the issue says, the Python client throws `Exception`s instead of a 
subclass of it (`PulsarException`), so the clients must catch the blanket 
Exception.
    
    ### Modifications
    
    Every C `PulsarException` is now thrown in Python with the same type.
---
 pulsar-client-cpp/python/CMakeLists.txt       |   3 +-
 pulsar-client-cpp/python/pulsar/__init__.py   |   2 +
 pulsar-client-cpp/python/pulsar/exceptions.py |  28 +++++
 pulsar-client-cpp/python/pulsar_test.py       | 171 +++++++++-----------------
 pulsar-client-cpp/python/src/enums.cc         |  13 ++
 pulsar-client-cpp/python/src/exceptions.cc    |  92 ++++++++++++++
 pulsar-client-cpp/python/src/pulsar.cc        |   6 +-
 7 files changed, 201 insertions(+), 114 deletions(-)

diff --git a/pulsar-client-cpp/python/CMakeLists.txt 
b/pulsar-client-cpp/python/CMakeLists.txt
index e78d80a..c110d01 100644
--- a/pulsar-client-cpp/python/CMakeLists.txt
+++ b/pulsar-client-cpp/python/CMakeLists.txt
@@ -29,7 +29,8 @@ ADD_LIBRARY(_pulsar SHARED src/pulsar.cc
                            src/authentication.cc
                            src/reader.cc
                            src/schema.cc
-                           src/cryptoKeyReader.cc)
+                           src/cryptoKeyReader.cc
+                           src/exceptions.cc)
 
 SET(CMAKE_SHARED_LIBRARY_PREFIX )
 SET(CMAKE_SHARED_LIBRARY_SUFFIX .so)
diff --git a/pulsar-client-cpp/python/pulsar/__init__.py 
b/pulsar-client-cpp/python/pulsar/__init__.py
index b47c87d..3f00554 100644
--- a/pulsar-client-cpp/python/pulsar/__init__.py
+++ b/pulsar-client-cpp/python/pulsar/__init__.py
@@ -103,6 +103,8 @@ import _pulsar
 
 from _pulsar import Result, CompressionType, ConsumerType, InitialPosition, 
PartitionsRoutingMode, BatchingType  # noqa: F401
 
+from pulsar.exceptions import *
+
 from pulsar.functions.function import Function
 from pulsar.functions.context import Context
 from pulsar.functions.serde import SerDe, IdentitySerDe, PickleSerDe
diff --git a/pulsar-client-cpp/python/pulsar/exceptions.py 
b/pulsar-client-cpp/python/pulsar/exceptions.py
new file mode 100644
index 0000000..d151564
--- /dev/null
+++ b/pulsar-client-cpp/python/pulsar/exceptions.py
@@ -0,0 +1,28 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from _pulsar import PulsarException, UnknownError, InvalidConfiguration, 
Timeout, LookupError, ConnectError, \
+    ReadError, AuthenticationError, AuthorizationError, 
ErrorGettingAuthenticationData, BrokerMetadataError, \
+    BrokerPersistenceError, ChecksumError, ConsumerBusy, NotConnected, 
AlreadyClosed, InvalidMessage, \
+    ConsumerNotInitialized, ProducerNotInitialized, ProducerBusy, 
TooManyLookupRequestException, InvalidTopicName, \
+    InvalidUrl, ServiceUnitNotReady, OperationNotSupported, 
ProducerBlockedQuotaExceededError, \
+    ProducerBlockedQuotaExceededException, ProducerQueueIsFull, MessageTooBig, 
TopicNotFound, SubscriptionNotFound, \
+    ConsumerNotFound, UnsupportedVersionError, TopicTerminated, CryptoError, 
IncompatibleSchema, ConsumerAssignError, \
+    CumulativeAcknowledgementNotAllowedError, 
TransactionCoordinatorNotFoundError, InvalidTxnStatusError, \
+    NotAllowedError, TransactionConflict, TransactionNotFound, ProducerFenced, 
MemoryBufferIsFull
diff --git a/pulsar-client-cpp/python/pulsar_test.py 
b/pulsar-client-cpp/python/pulsar_test.py
index e7d05f3..71b67d7 100755
--- a/pulsar-client-cpp/python/pulsar_test.py
+++ b/pulsar-client-cpp/python/pulsar_test.py
@@ -42,6 +42,7 @@ except ImportError:
 
 TM = 10000  # Do not wait forever in tests
 
+
 def doHttpPost(url, data):
     req = Request(url, data.encode())
     req.add_header('Content-Type', 'application/json')
@@ -67,6 +68,7 @@ def doHttpGet(url):
     req.add_header('Accept', 'application/json')
     return urlopen(req).read()
 
+
 class PulsarTest(TestCase):
 
     serviceUrl = 'pulsar://localhost:6650'
@@ -96,6 +98,16 @@ class PulsarTest(TestCase):
         conf.consumer_name("my-name")
         self.assertEqual(conf.consumer_name(), "my-name")
 
+    def test_connect_error(self):
+        with self.assertRaises(pulsar.ConnectError):
+            client = Client('fakeServiceUrl')
+            client.create_producer('connect-error-topic')
+            client.close()
+
+    def test_exception_inheritance(self):
+        assert issubclass(pulsar.ConnectError, pulsar.PulsarException)
+        assert issubclass(pulsar.PulsarException, Exception)
+
     def test_simple_producer(self):
         client = Client(self.serviceUrl)
         producer = client.create_producer('my-python-topic')
@@ -147,11 +159,8 @@ class PulsarTest(TestCase):
         self.assertTrue(msg)
         self.assertEqual(msg.data(), b'hello')
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         consumer.unsubscribe()
         client.close()
@@ -189,11 +198,8 @@ class PulsarTest(TestCase):
         producer.send(b'hello', deliver_at=int(round(time.time() * 1000)) + 
1100)
 
         # Message should not be available in the next second
-        try:
-            msg = consumer.receive(1000)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(1000)
 
         # Message should be published now
         msg = consumer.receive(TM)
@@ -213,11 +219,8 @@ class PulsarTest(TestCase):
         producer.send(b'hello', deliver_after=timedelta(milliseconds=1100))
 
         # Message should not be available in the next second
-        try:
-            msg = consumer.receive(1000)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(1000)
 
         # Message should be published in the next 500ms
         msg = consumer.receive(TM)
@@ -229,14 +232,14 @@ class PulsarTest(TestCase):
 
     def test_consumer_initial_position(self):
         client = Client(self.serviceUrl)
-        producer = client.create_producer('my-python-topic-producer-consumer')
+        producer = client.create_producer('consumer-initial-position')
 
         # Sending 5 messages before consumer creation.
         # These should be received with initial_position set to Earliest but 
not with Latest.
         for i in range(5):
             producer.send(b'hello-%d' % i)
 
-        consumer = client.subscribe('my-python-topic-producer-consumer',
+        consumer = client.subscribe('consumer-initial-position',
                                     'my-sub',
                                     consumer_type=ConsumerType.Shared,
                                     initial_position=InitialPosition.Earliest)
@@ -250,11 +253,8 @@ class PulsarTest(TestCase):
             self.assertTrue(msg)
             self.assertEqual(msg.data(), b'hello-%d' % i)
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         consumer.unsubscribe()
         client.close()
@@ -310,21 +310,18 @@ class PulsarTest(TestCase):
                         tls_allow_insecure_connection=False,
                         authentication=AuthenticationTLS(certs_dir + 
'client-cert.pem', certs_dir + 'client-key.pem'))
 
-        consumer = client.subscribe('my-python-topic-producer-consumer',
+        consumer = client.subscribe('my-python-topic-tls-auth',
                                     'my-sub',
                                     consumer_type=ConsumerType.Shared)
-        producer = client.create_producer('my-python-topic-producer-consumer')
+        producer = client.create_producer('my-python-topic-tls-auth')
         producer.send(b'hello')
 
         msg = consumer.receive(TM)
         self.assertTrue(msg)
         self.assertEqual(msg.data(), b'hello')
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         client.close()
 
@@ -340,21 +337,18 @@ class PulsarTest(TestCase):
                         tls_allow_insecure_connection=False,
                         authentication=Authentication(authPlugin, authParams))
 
-        consumer = client.subscribe('my-python-topic-producer-consumer',
+        consumer = client.subscribe('my-python-topic-tls-auth-2',
                                     'my-sub',
                                     consumer_type=ConsumerType.Shared)
-        producer = client.create_producer('my-python-topic-producer-consumer')
+        producer = client.create_producer('my-python-topic-tls-auth-2')
         producer.send(b'hello')
 
         msg = consumer.receive(TM)
         self.assertTrue(msg)
         self.assertEqual(msg.data(), b'hello')
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         client.close()
 
@@ -389,21 +383,18 @@ class PulsarTest(TestCase):
                         tls_allow_insecure_connection=False,
                         authentication=Authentication(authPlugin, authParams))
 
-        consumer = client.subscribe('my-python-topic-producer-consumer',
+        consumer = client.subscribe('my-python-topic-tls-auth-3',
                                     'my-sub',
                                     consumer_type=ConsumerType.Shared)
-        producer = client.create_producer('my-python-topic-producer-consumer')
+        producer = client.create_producer('my-python-topic-tls-auth-3')
         producer.send(b'hello')
 
         msg = consumer.receive(TM)
         self.assertTrue(msg)
         self.assertEqual(msg.data(), b'hello')
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         client.close()
 
@@ -417,12 +408,11 @@ class PulsarTest(TestCase):
                         tls_trust_certs_file_path=certs_dir + 'cacert.pem',
                         tls_allow_insecure_connection=False,
                         authentication=Authentication(authPlugin, authParams))
-        try:
-            client.subscribe('my-python-topic-producer-consumer',
+
+        with self.assertRaises(pulsar.ConnectError):
+            client.subscribe('my-python-topic-auth-junk-params',
                              'my-sub',
                              consumer_type=ConsumerType.Shared)
-        except:
-            pass  # Exception is expected
 
     def test_message_listener(self):
         client = Client(self.serviceUrl)
@@ -462,11 +452,8 @@ class PulsarTest(TestCase):
         self.assertTrue(msg)
         self.assertEqual(msg.data(), b'hello')
 
-        try:
-            msg = reader.read_next(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            reader.read_next(100)
 
         reader.close()
         client.close()
@@ -618,13 +605,8 @@ class PulsarTest(TestCase):
             self.assertEqual(msg.data(), b'hello-%d' % i)
             consumer.acknowledge(msg)
 
-        try:
-            # No other messages should be received
-            consumer.receive(timeout_millis=1000)
-            self.assertTrue(False)
-        except:
-            # Exception is expected
-            pass
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         producer.close()
 
@@ -636,13 +618,8 @@ class PulsarTest(TestCase):
         producer.send(b'hello-2', sequence_id=2)
         self.assertEqual(producer.last_sequence_id(), 2)
 
-        try:
-            # No other messages should be received
-            consumer.receive(timeout_millis=1000)
-            self.assertTrue(False)
-        except:
-            # Exception is expected
-            pass
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         doHttpPost(self.adminUrl + 
'/admin/v2/namespaces/public/default/deduplication',
                    'false')
@@ -873,11 +850,8 @@ class PulsarTest(TestCase):
 
         # repeat with reader
         reader = client.create_reader('my-python-topic-seek', MessageId.latest)
-        try:
-            msg = reader.read_next(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            reader.read_next(100)
 
         # earliest
         reader.seek(MessageId.earliest)
@@ -925,11 +899,8 @@ class PulsarTest(TestCase):
         self.assertEqual(msg.data(), b'hello')
         consumer.acknowledge(msg)
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         client.close()
 
@@ -972,13 +943,8 @@ class PulsarTest(TestCase):
             msg = consumer.receive(TM)
             consumer.acknowledge(msg)
 
-        try:
-        # No other messages should be received
-            consumer.receive(timeout_millis=500)
-            self.assertTrue(False)
-        except:
-            # Exception is expected
-            pass
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
         client.close()
 
     def test_topics_pattern_consumer(self):
@@ -1027,13 +993,8 @@ class PulsarTest(TestCase):
             msg = consumer.receive(TM)
             consumer.acknowledge(msg)
 
-        try:
-            # No other messages should be received
-            consumer.receive(timeout_millis=500)
-            self.assertTrue(False)
-        except:
-            # Exception is expected
-            pass
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
         client.close()
 
     def test_message_id(self):
@@ -1110,11 +1071,8 @@ class PulsarTest(TestCase):
         self.assertTrue(msg)
         self.assertEqual(msg.data(), b'hello')
 
-        try:
-            msg = consumer.receive(100)
-            self.assertTrue(False)  # Should not reach this point
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
 
         consumer.unsubscribe()
         client.close()
@@ -1158,7 +1116,8 @@ class PulsarTest(TestCase):
         client = Client(self.serviceUrl)
         consumer = client.subscribe('test_negative_acks',
                                     'test',
-                                    schema=pulsar.schema.StringSchema())
+                                    schema=pulsar.schema.StringSchema(),
+                                    negative_ack_redelivery_delay_ms=1000)
         producer = client.create_producer('test_negative_acks',
                                           schema=pulsar.schema.StringSchema())
         for i in range(10):
@@ -1176,29 +1135,17 @@ class PulsarTest(TestCase):
             self.assertEqual(msg.value(), "hello-%d" % i)
             consumer.acknowledge(msg)
 
-        try:
-            # No more messages expected
-            msg = consumer.receive(100)
-            self.assertTrue(False)
-        except:
-            pass  # Exception is expected
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
         client.close()
 
     def _check_value_error(self, fun):
-        try:
+        with self.assertRaises(ValueError):
             fun()
-            # Should throw exception
-            self.assertTrue(False)
-        except ValueError:
-            pass  # Expected
 
     def _check_type_error(self, fun):
-        try:
+        with self.assertRaises(TypeError):
             fun()
-            # Should throw exception
-            self.assertTrue(False)
-        except TypeError:
-            pass  # Expected
 
 
 if __name__ == '__main__':
diff --git a/pulsar-client-cpp/python/src/enums.cc 
b/pulsar-client-cpp/python/src/enums.cc
index f57e08d..c23b211 100644
--- a/pulsar-client-cpp/python/src/enums.cc
+++ b/pulsar-client-cpp/python/src/enums.cc
@@ -63,6 +63,7 @@ void export_enums() {
             .value("InvalidMessage", ResultInvalidMessage)
             .value("ConsumerNotInitialized", ResultConsumerNotInitialized)
             .value("ProducerNotInitialized", ResultProducerNotInitialized)
+            .value("ProducerBusy", ResultProducerBusy)
             .value("TooManyLookupRequestException", 
ResultTooManyLookupRequestException)
             .value("InvalidTopicName", ResultInvalidTopicName)
             .value("InvalidUrl", ResultInvalidUrl)
@@ -76,6 +77,18 @@ void export_enums() {
             .value("SubscriptionNotFound", ResultSubscriptionNotFound)
             .value("ConsumerNotFound", ResultConsumerNotFound)
             .value("UnsupportedVersionError", ResultUnsupportedVersionError)
+            .value("TopicTerminated", ResultTopicTerminated)
+            .value("CryptoError", ResultCryptoError)
+            .value("IncompatibleSchema", ResultIncompatibleSchema)
+            .value("ConsumerAssignError", ResultConsumerAssignError)
+            .value("CumulativeAcknowledgementNotAllowedError", 
ResultCumulativeAcknowledgementNotAllowedError)
+            .value("TransactionCoordinatorNotFoundError", 
ResultTransactionCoordinatorNotFoundError)
+            .value("InvalidTxnStatusError", ResultInvalidTxnStatusError)
+            .value("NotAllowedError", ResultNotAllowedError)
+            .value("TransactionConflict", ResultTransactionConflict)
+            .value("TransactionNotFound", ResultTransactionNotFound)
+            .value("ProducerFenced", ResultProducerFenced)
+            .value("MemoryBufferIsFull", ResultMemoryBufferIsFull)
             ;
 
     enum_<SchemaType>("SchemaType", "Supported schema types")
diff --git a/pulsar-client-cpp/python/src/exceptions.cc 
b/pulsar-client-cpp/python/src/exceptions.cc
new file mode 100644
index 0000000..c39b52d
--- /dev/null
+++ b/pulsar-client-cpp/python/src/exceptions.cc
@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include <map>
+
+#include "utils.h"
+
+static PyObject* basePulsarException = nullptr;
+std::map<Result, PyObject*> exceptions;
+
+PyObject* createExceptionClass(const char* name, PyObject* baseTypeObj = 
PyExc_Exception) {
+    using namespace boost::python;
+
+    std::string fullName = "_pulsar.";
+    fullName += name;
+
+    PyObject* typeObj = PyErr_NewException(const_cast<char*>(fullName.c_str()),
+                                           baseTypeObj, nullptr);
+    if (!typeObj) throw_error_already_set();
+    scope().attr(name) = handle<>(borrowed(typeObj));
+    return typeObj;
+}
+
+PyObject* get_exception_class(Result result) {
+    return exceptions[result];
+}
+
+void export_exceptions() {
+    using namespace boost::python;
+
+    basePulsarException = createExceptionClass("PulsarException");
+
+    exceptions[ResultUnknownError] = createExceptionClass("UnknownError", 
basePulsarException);
+    exceptions[ResultInvalidConfiguration] = 
createExceptionClass("InvalidConfiguration", basePulsarException);
+    exceptions[ResultTimeout] = createExceptionClass("Timeout", 
basePulsarException);
+    exceptions[ResultLookupError] = createExceptionClass("LookupError", 
basePulsarException);
+    exceptions[ResultConnectError] = createExceptionClass("ConnectError", 
basePulsarException);
+    exceptions[ResultReadError] = createExceptionClass("ReadError", 
basePulsarException);
+    exceptions[ResultAuthenticationError] = 
createExceptionClass("AuthenticationError", basePulsarException);
+    exceptions[ResultAuthorizationError] = 
createExceptionClass("AuthorizationError", basePulsarException);
+    exceptions[ResultErrorGettingAuthenticationData] = 
createExceptionClass("ErrorGettingAuthenticationData", basePulsarException);
+    exceptions[ResultBrokerMetadataError] = 
createExceptionClass("BrokerMetadataError", basePulsarException);
+    exceptions[ResultBrokerPersistenceError] = 
createExceptionClass("BrokerPersistenceError", basePulsarException);
+    exceptions[ResultChecksumError] = createExceptionClass("ChecksumError", 
basePulsarException);
+    exceptions[ResultConsumerBusy] = createExceptionClass("ConsumerBusy", 
basePulsarException);
+    exceptions[ResultNotConnected] = createExceptionClass("NotConnected", 
basePulsarException);
+    exceptions[ResultAlreadyClosed] = createExceptionClass("AlreadyClosed", 
basePulsarException);
+    exceptions[ResultInvalidMessage] = createExceptionClass("InvalidMessage", 
basePulsarException);
+    exceptions[ResultConsumerNotInitialized] = 
createExceptionClass("ConsumerNotInitialized", basePulsarException);
+    exceptions[ResultProducerNotInitialized] = 
createExceptionClass("ProducerNotInitialized", basePulsarException);
+    exceptions[ResultProducerBusy] = createExceptionClass("ProducerBusy", 
basePulsarException);
+    exceptions[ResultTooManyLookupRequestException] = 
createExceptionClass("TooManyLookupRequestException", basePulsarException);
+    exceptions[ResultInvalidTopicName] = 
createExceptionClass("InvalidTopicName", basePulsarException);
+    exceptions[ResultInvalidUrl] = createExceptionClass("InvalidUrl", 
basePulsarException);
+    exceptions[ResultServiceUnitNotReady] = 
createExceptionClass("ServiceUnitNotReady", basePulsarException);
+    exceptions[ResultOperationNotSupported] = 
createExceptionClass("OperationNotSupported", basePulsarException);
+    exceptions[ResultProducerBlockedQuotaExceededError] = 
createExceptionClass("ProducerBlockedQuotaExceededError", basePulsarException);
+    exceptions[ResultProducerBlockedQuotaExceededException] = 
createExceptionClass("ProducerBlockedQuotaExceededException", 
basePulsarException);
+    exceptions[ResultProducerQueueIsFull] = 
createExceptionClass("ProducerQueueIsFull", basePulsarException);
+    exceptions[ResultMessageTooBig] = createExceptionClass("MessageTooBig", 
basePulsarException);
+    exceptions[ResultTopicNotFound] = createExceptionClass("TopicNotFound", 
basePulsarException);
+    exceptions[ResultSubscriptionNotFound] = 
createExceptionClass("SubscriptionNotFound", basePulsarException);
+    exceptions[ResultConsumerNotFound] = 
createExceptionClass("ConsumerNotFound", basePulsarException);
+    exceptions[ResultUnsupportedVersionError] = 
createExceptionClass("UnsupportedVersionError", basePulsarException);
+    exceptions[ResultTopicTerminated] = 
createExceptionClass("TopicTerminated", basePulsarException);
+    exceptions[ResultCryptoError] = createExceptionClass("CryptoError", 
basePulsarException);
+    exceptions[ResultIncompatibleSchema] = 
createExceptionClass("IncompatibleSchema", basePulsarException);
+    exceptions[ResultConsumerAssignError] = 
createExceptionClass("ConsumerAssignError", basePulsarException);
+    exceptions[ResultCumulativeAcknowledgementNotAllowedError] = 
createExceptionClass("CumulativeAcknowledgementNotAllowedError", 
basePulsarException);
+    exceptions[ResultTransactionCoordinatorNotFoundError] = 
createExceptionClass("TransactionCoordinatorNotFoundError", 
basePulsarException);
+    exceptions[ResultInvalidTxnStatusError] = 
createExceptionClass("InvalidTxnStatusError", basePulsarException);
+    exceptions[ResultNotAllowedError] = 
createExceptionClass("NotAllowedError", basePulsarException);
+    exceptions[ResultTransactionConflict] = 
createExceptionClass("TransactionConflict", basePulsarException);
+    exceptions[ResultTransactionNotFound] = 
createExceptionClass("TransactionNotFound", basePulsarException);
+    exceptions[ResultProducerFenced] = createExceptionClass("ProducerFenced", 
basePulsarException);
+    exceptions[ResultMemoryBufferIsFull] = 
createExceptionClass("MemoryBufferIsFull", basePulsarException);
+}
diff --git a/pulsar-client-cpp/python/src/pulsar.cc 
b/pulsar-client-cpp/python/src/pulsar.cc
index f80c9a4..a46ce53 100644
--- a/pulsar-client-cpp/python/src/pulsar.cc
+++ b/pulsar-client-cpp/python/src/pulsar.cc
@@ -28,13 +28,16 @@ void export_enums();
 void export_authentication();
 void export_schema();
 void export_cryptoKeyReader();
+void export_exceptions();
+
+PyObject* get_exception_class(Result result);
 
 
 static void translateException(const PulsarException& ex) {
     std::string err = "Pulsar error: ";
     err += strResult(ex._result);
 
-    PyErr_SetString(PyExc_Exception, err.c_str());
+    PyErr_SetString(get_exception_class(ex._result), err.c_str());
 }
 
 BOOST_PYTHON_MODULE(_pulsar)
@@ -55,4 +58,5 @@ BOOST_PYTHON_MODULE(_pulsar)
     export_authentication();
     export_schema();
     export_cryptoKeyReader();
+    export_exceptions();
 }

Reply via email to