[fix][client] Fix MessageIdUtils cannot handle TopicMessageId

[RobertIndie]

Motivation

PIP-229 introduces an interface change for the messageId which break the behavior in and after Pulsar 3.0.

This leads to the MessageIdUtils not be able to handle the TopicMessageId. It will throw error like:

java.lang.ClassCastException: class org.apache.pulsar.client.impl.TopicMessageIdImpl cannot be cast to class org.apache.pulsar.client.impl.MessageIdImpl (org.apache.pulsar.client.impl.TopicMessageIdImpl and org.apache.pulsar.client.impl.MessageIdImpl are in unnamed module of loader 'app')

	at org.apache.pulsar.client.util.MessageIdUtils.getOffset(MessageIdUtils.java:27)
...

Modifications

• Use MessageIdAdv to convert the MessageId to offset

Verifying this change

This change added tests.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[BewareMyPower]

This util class was introduced at very early era. It's better to add the util function to MessageIdAdvUtils for MessageIdAdv.

And I don't think it's worth providing a util function for MessageIdAdv in the Pulsar core repo. MessageIdAdv is introduced mainly for this reason. The downstream application should cast MessageId to MessageIdAdv to access the internal fields. It should not depend on the wrapped APIs from the upstream.

Pulsar does not have the "offset" concept, the same logic is only used in the Kafka sink connector, but it does not call this util method directly. See

pulsar/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java

Line 376 in e558cfe

long offset = (ledgerId << 28) | entryId;

To get the offset, it's better to leverage the AppendIndexMetadataInterceptor like KoP. See the discussion here: streamnative/kop#290

[RobertIndie]

This is also used in third-party connectors, not just limited to the kafka sink connector.
Removing the offset from the pulsar-client looks good to me. I think we can mark it with Deprecate.

[BewareMyPower]

Yes. Before removing it, we can mark it with @Deprecated.

In 3rd party applications, it's easy to customize its own implementation via MessageIdAdv. You can see the example from an early implementation of KoP: https://github.com/streamnative/kop/blob/branch-2.7.4.5/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageIdUtils.java, which is also an example to show the way to compute the offset is not standard.

[RobertIndie]

To leave our discussion context here, I created a new PR to deprecate these APIs: #22747.
There is the same offset implementation in the FunctionCommon. I don't deprecate them currently because the connector still needs to use it.