Python pulsar-client reader causes unidentifiable C++ exception

[alphara]

I am using Python pulsar-client. I create a reader that starts to read message from earliest.

msg_id = MessageId.earliest
reader = client.create_reader('my-topic', msg_id)

I am getting an error message

Connect error: unidentifiable C++ exception

Please, advice.

[sijie]

@alphara what version of pulsar python client are you using? Did you install the same version of Pulsar CPP client?

[alphara]

I am using lib-pulsar v2.6.0 and pulsar-client>=2.5.1 in Python app.

[alphara]

I just tried lib-pulsar v2.6.0 and pulsar-client v2.6.0. I see the same C++ exception.

[merlimat]

The Python client is not using the C++ libpulsar, rather it's embedded in the wheel file itself.

@alphara Can you provide a simple reproducer example? Also, with the complete original logs?

For example, this is generally working fine with a connect error:

In [1]: import pulsar

In [2]: c = pulsar.Client('pulsar://localhost:6650')

In [3]: r = c.create_reader('my-topic', pulsar.MessageId.earliest)
2020-06-23 15:28:12.479 INFO  ConnectionPool:85 | Created connection for pulsar://localhost:6650
2020-06-23 15:28:12.481 ERROR ClientConnection:380 | [<none> -> pulsar://localhost:6650] Failed to establish connection: Connection refused
2020-06-23 15:28:12.481 INFO  ClientConnection:1349 | [<none> -> pulsar://localhost:6650] Connection closed
2020-06-23 15:28:12.481 ERROR ClientImpl:219 | Error Checking/Getting Partition Metadata while creating readeron persistent://public/default/my-topic -- 5
2020-06-23 15:28:12.481 INFO  ClientConnection:235 | [<none> -> pulsar://localhost:6650] Destroyed connection

[alphara]

The exception arises after the reader gets connected. Try to launch Pulsar before you launch the sample code.

The pulsar docs says here:

Pulsar Python client library is a wrapper over the existing C++ client library and exposes all of the same features. You can find the code in the python subdirectory of the C++ client code.

Is this info obsolete?

This is a simple example that I use

from pulsar import Client, MessageId

client = Client(service_url='pulsar://localhost:6650',
                authentication=None)
topic = 'my-topic'
msg_id = MessageId.earliest
reader = client.create_reader(topic, msg_id)

while True:
    msg = reader.read_next()
    print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))

I found that the example above works on my machine.

I have a bigger application that uses asyncio library. The reader.read_next() is executed inside the run_in_executor() method.

  # ...

  async def _tryToSubscribe(self, topic, queue, callback, pulsarStartMessage):
    def sync_subscribe():
      startMessageId = MessageId.earliest
      reader = self.pulsar.create_reader(topic, startMessageId)     # <- exception occurs here

    await self.loop.run_in_executor(self.executor, sync_subscribe)
    # ...

[merlimat]

Is this info obsolete?

No, the Python client is based on C++. What I meant is that you don't need to install libpulsar separately. When you do pip install pulsar-client it will be fetching a wheel file that contains everything that is needed.

I have a bigger application that uses asyncio library. The reader.read_next() is executed inside the run_in_executor() method.

Can you post a small repro code using asyncio?
Can you show the pulsar client lib logs?

[alphara]

Here is a small example of code using asyncio.

from asyncio import get_event_loop, get_running_loop, sleep
from asyncio import CancelledError, Task
from signal import SIGINT, SIGTERM
from pulsar import Client, MessageId


async def start():
  loop = get_running_loop()
  executor = None

  def sync_connect():
    client = Client(service_url='pulsar://localhost:6650',
                    authentication=None)
    return client

  client = await loop.run_in_executor(executor, sync_connect)

  def sync_subscribe():
    topic = 'my-topic-1'
    msg_id = MessageId.earliest
    reader = client.create_reader(topic, msg_id)
    return reader

  reader = await loop.run_in_executor(executor, sync_subscribe)

  def sync_receive():
    msg = reader.read_next()
    return msg

  while True:
    msg = await loop.run_in_executor(executor, sync_receive)
    print("Received message '{}' id='{}'".format(msg.data(),
                                                 msg.message_id()))

  client.close()


async def run():
  ''' Run in event loop '''
  await start()

  while True:
    try:
      await sleep(60)
    except CancelledError:
      break
  get_running_loop().stop()


def signal_handler():
  ''' Signal handler '''
  for task in Task.all_tasks():
    task.cancel()


def main():
  ''' Program entry point '''
  loop = get_event_loop()
  loop.add_signal_handler(SIGINT, signal_handler)
  loop.add_signal_handler(SIGTERM, signal_handler)
  loop.run_until_complete(run())
  loop.remove_signal_handler(SIGINT)
  loop.remove_signal_handler(SIGTERM)
  loop.close()


if __name__ == '__main__':
  main()

[merlimat]

It seems to be working for me. Using your code above with Python3.8 and pulsar-client==2.6.0.

$ python3 a.py
2020-06-26 13:56:55.615 INFO  [0x70000321d000] ConnectionPool:85 | Created connection for pulsar://localhost:6650
2020-06-26 13:56:55.617 INFO  [0x7000032a0000] ClientConnection:335 | [127.0.0.1:65389 -> 127.0.0.1:6650] Connected to broker
2020-06-26 13:56:55.621 INFO  [0x7000032a0000] AckGroupingTrackerEnabled:53 | ACK grouping is enabled, grouping time 100ms, grouping max size 1000
2020-06-26 13:56:55.621 INFO  [0x7000032a0000] HandlerBase:53 | [persistent://public/default/my-topic-1, reader-06cf831e5f, 0] Getting connection from pool
2020-06-26 13:56:55.622 INFO  [0x7000032a0000] ConsumerImpl:199 | [persistent://public/default/my-topic-1, reader-06cf831e5f, 0] Created consumer on broker [127.0.0.1:65389 -> 127.0.0.1:6650]

[alphara]

@merlimat, thank you for testing. Good to know it works on your side.

[merlimat]

Good to know it works on your side.

Well.. it would be better if it actually failed 😄