Pulsar: [ClientAPI]ReaderBuilder.loadConf() not working

[yjshen]

Describe the bug
Cannot create new reader while using loadConf method in reader builder chain.

To Reproduce
Steps to reproduce the behavior:

1. Create PulsarClient
2. client.newReader().topic(tp).startMessageId(start).loadConf(readerConf).create()

Expected behavior
Create a reader with both settings specified using methods as well as loadConf.

Screenshots

java.lang.RuntimeException: Failed to load config into existing configuration data
.....
Caused by: org.apache.pulsar.shade.com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `org.apache.pulsar.client.api.MessageId` (no Creators, like default construct, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information
 at [Source: (String)"{"cryptoFailureAction":"FAIL","topicName":"topic-0","readCompacted":false,"startMessageId":{"ledgerId":-1,"entryId":-1,"partitionIndex":-1},"receiverQueueSize":1000}"; line: 1, column: 92] (through reference chain: org.apache.pulsar.client.impl.conf.ReaderConfigurationData["startMessageId"])
    at org.apache.pulsar.shade.com.fasterxml.jackson.databind.exc.InvalidDefinitionException.from(InvalidDefinitionException.java:67) ~[pulsar-client-all-2.4.0-c509f86ba.jar:2.4.0-c509f86ba]

Desktop (please complete the following information):

• OS: MacOS

Additional context
This may caused by ReaderConfigurationData contains a startMessageId of MessageId field that cannot be constructed.
As abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information implies.

[yjshen]

@sijie @jiazhai

[ambition119]

@yjshen I can't reproduce this issue, what about example code ?

public static void main(String[] args) throws Exception {
    try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build()) {
      Map<String, Object> config = new HashMap<>();
      config.put("topicName", "test_src");
      config.put("receiverQueueSize", 2000);
      MessageId id = new MessageIdImpl(1, 2, 3);
      ReaderBuilder<byte[]> builder = client.newReader().topic("test_src").loadConf(config)
          .startMessageId(id);   //run ok
//          or 
//      .startMessageId(MessageId.earliest);  // run ok
      Reader<byte[]> reader = builder.create();
      if (reader.hasMessageAvailable()) {
        System.out.println(new String(reader.readNext().getData()));
      }
    }
  }

[yjshen]

Put startMessageId before loadConf in the builder chain could reproduce it.

[ambition119]

Put startMessageId before loadConf in the builder chain could reproduce it.

thanks, I can try solve.

[sijie]

@ambition119 any luck?

[ambition119]

It was revised on Friday and will be submitted PR today, then you can have a look, thanks.