python API missing `is_read_compacted` in create_reader()

[candlerb]

Describe the bug

The python API has is_read_compacted in the Consumer API, but not in the Reader API.

I think this may be an accidental omission. I checked the Java API docs, both ConsumerBuilder and ReaderBuilder have readCompacted

Similarly, the C++ API has setReadCompacted in ReaderConfiguration.cc

To Reproduce

import pulsar

client = pulsar.Client('pulsar://localhost:6650')

msg_id = pulsar.MessageId.earliest

reader = client.create_reader('my-topic', msg_id, is_read_compacted=True)

while True:
    msg = reader.read_next()
    print("Received message %r id=%s" % (msg.data(), msg.message_id()))

Gives TypeError: create_reader() got an unexpected keyword argument 'is_read_compacted'

Expected behavior

To work the same as the consumer API. The following works:

consumer = client.subscribe('my-topic', 'my-subscription', is_read_compacted=True)

Desktop (please complete the following information):

• OS: Ubuntu 18.04, Pulsar 2.4.1, python 3.6.8, and pulsar-client==2.4.1

[jiazhai]

will look into it.

[jiazhai]

@candlerb , have you tried read_compacted? it is defined as this in the code:

    class_<ReaderConfiguration>("ReaderConfiguration")
            .def("message_listener", &ReaderConfiguration_setReaderListener, return_self<>())
            .def("schema", &ReaderConfiguration::getSchema, return_value_policy<copy_const_reference>())
            .def("schema", &ReaderConfiguration::setSchema, return_self<>())
            .def("receiver_queue_size", &ReaderConfiguration::getReceiverQueueSize)
            .def("receiver_queue_size", &ReaderConfiguration::setReceiverQueueSize)
            .def("reader_name", &ReaderConfiguration::getReaderName, return_value_policy<copy_const_reference>())
            .def("reader_name", &ReaderConfiguration::setReaderName)
            .def("subscription_role_prefix", &ReaderConfiguration::getSubscriptionRolePrefix, return_value_policy<copy_const_reference>())
            .def("subscription_role_prefix", &ReaderConfiguration::setSubscriptionRolePrefix)
            .def("read_compacted", &ReaderConfiguration::isReadCompacted)
            .def("read_compacted", &ReaderConfiguration::setReadCompacted)

[candlerb]

Yes I had already tried it, and it fails.

Code:

import pulsar

client = pulsar.Client('pulsar://localhost:6650')

msg_id = pulsar.MessageId.earliest

reader = client.create_reader('my-topic', msg_id, read_compacted=True)

while True:
    msg = reader.read_next()
    print("Received message %r id=%s" % (msg.data(), msg.message_id()))

Exception:

$ python3 read.py
Traceback (most recent call last):
  File "read.py", line 7, in <module>
    reader = client.create_reader('my-topic', msg_id, read_compacted=True)
TypeError: create_reader() got an unexpected keyword argument 'read_compacted'

This is with pulsar-client (2.4.1)

[candlerb]

I notice in pulsar-client-cpp/python/pulsar/__init__.py that subscribe has this option:

    def subscribe(self, topic, subscription_name,
                  consumer_type=ConsumerType.Exclusive,
                  schema=schema.BytesSchema(),
                  message_listener=None,
                  receiver_queue_size=1000,
                  max_total_receiver_queue_size_across_partitions=50000,
                  consumer_name=None,
                  unacked_messages_timeout_ms=None,
                  broker_consumer_stats_cache_time_ms=30000,
                  negative_ack_redelivery_delay_ms=60000,
                  is_read_compacted=False,   <<<<<
                  properties=None,
                  pattern_auto_discovery_period=60,
                  initial_position=InitialPosition.Latest
                  ):
...
        conf = _pulsar.ConsumerConfiguration()
        conf.consumer_type(consumer_type)
        conf.read_compacted(is_read_compacted)

but create_reader does not:

    def create_reader(self, topic, start_message_id,
                      schema=schema.BytesSchema(),
                      reader_listener=None,
                      receiver_queue_size=1000,
                      reader_name=None,
                      subscription_role_prefix=None
                      ):
...
        conf = _pulsar.ReaderConfiguration()

I think it's just omitting to pass this on.

[candlerb]

I made a fix, it's pretty simple (PR to follow). However I spotted another problem:

        if reader_listener:
            conf.reader_listener(_listener_wrapper(reader_listener, schema))

In the code you posted before, should

            .def("message_listener", &ReaderConfiguration_setReaderListener, return_self<>())

be changed to

            .def("reader_listener", &ReaderConfiguration_setReaderListener, return_self<>())

? Or should the python code be changed to use message_listener instead? EDIT: reproduced and raised separately as #5484

[jiazhai]

Thanks @candlerb for the verify.
And Yes, Your change is needed.

            .def("reader_listener", &ReaderConfiguration_setReaderListener, return_self<>())

👍