[improve][client] PIP-229: Add a common interface to get fields of MessageIdData

[BewareMyPower]

Master issue: #18950

Motivation

We need a common interface to get fields of the MessageIdData. After that, we won't need to assert a MessageId implementation is an instance of a specific class. And we can pass our customized MessageId implementation to APIs like acknowledge and seek.

Modifications

• Add MessageIdAdv to get fields of MessageIdData, make all MessageId implementations inherit it (except MultiMessageIdImpl).
• Add MessageIdAdvUtils for the most common used methods.
• Replace BatchMessageAcker with the BitSet for ACK.
• Remove TopicMessageIdImpl#getInnerMessageId since a TopicMessageIdImpl can be treated as its underlying MessageId implementation now.
• Remove instanceof BatchMessageIdImpl checks in pulsar-client module by casting to MessageIdAdv.

After this refactoring, the 3rd party library will no longer need to cast a MessageId to a specific implementation. It only needs to cast MessageId to MessageIdAdv. Users can also implement their own util class so the methods of MessageIdAdvUtils are all not public.

Verifications

Add CustomMessageIdTest to verify a simple MessageIdAdv implementation that only has the (ledger id, entry id, batch idx, batch size) fields also works for seek and acknowledgment.

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#19

[BewareMyPower]

ping @congbobo184 @eolivelli @gaoran10 @codelipenghui @tisonkun @Demogorgon314 @Technoboy-

[rdhabalia]

@BewareMyPower this is a mess. We had discussed multiple times in Pulsar community to avoid any util class for MessageID because it hurts and makes complicated when we want to integrate any other components (eg: storm/spark adapter, Functions). This discussion happened many times and we rejected it.
I am not in favor of MessageIdAdvUtils and I don't think this is a good idea to implement in this way.

[rdhabalia]

oh.. this is wrong. we should not expose ledgerID/entryID in any public interface. we are opening a can of worms and we had prevented many times in Pulsar community. we can't check all the times for all the PRs but this PR must be reverted back to avoid exposing any internal details of Pulsar. It will be difficult in future to support this API if storage layer will not be bookie or bookie API will be changed.

[BewareMyPower]

This proposal has been discussed and modified into the current version of this PR in https://lists.apache.org/thread/25rzflmkfmvxhf3my0ombnbpv7bvgy32 and received 3 binding +1s and 2 non-binding +1s in https://lists.apache.org/thread/kmjq6lf1f11mf6qb8onhnlr17n27fcv4.

The PIP is here: #18950

As I've explained in the PIP:

These details might be not much useful to application users, but they are important to developers of Pulsar and its ecosystems.

It will be difficult in future to support this API if storage layer will not be bookie or bookie API will be changed.

Currently MessageId is still exposed to users, which does not expose these internal details. The new MessageIdAdv interface is only used for covenience. Users should write the following code after this PR:

MessageIdAdv msgIdAdv = (MessageIdAdv) msgId;
if (msgIdAdv != null) {
    // Get the ledger id, entry id directly
} else {
    // Handle the case when the storage layer might change in future
}

Before this PR, there is much code like:

MessageIdImpl impl = (MessageIdImpl) msgId;
if (impl != null) {
    // Check if msgId is another implementation...
} else {
    // ...
}

You can also see these code references in the changes of this PR. If you're going to revert this PIP, please at least start the discussion in the mail list to hear more voices.

[rdhabalia]

we should not let users use ledgerId/entryId. what if we replace bookkeeper with other some other storage? in that case, such codebase will be broken and exposing internal data and encouraging users to use it not a good design decision. I didn't review this PIP before but such discussion happened multiple times in past and Pulsar community had rejected this proposal in past. As we have not done any release yet, so, it's better to revert this PR soon.

[BewareMyPower]

we should not let users use ledgerId/entryId.

It's an ideal assumption. In real world, there are a lot of code references that use ledgerId/entryId from the MessageId. They have to hack into the pulsar-client module and cast MessageId to a specific implementation.

[BewareMyPower]

what if we replace bookkeeper with other some other storage?

When users need to access the detailed fields, they should have assumed the storage is BookKeeper. Otherwise, there is no need to get the storage details from the MessageId.

[BewareMyPower]

The fact is that Pulsar provides a hard-to-use interface MessageId. Then users, including ecosystem developers, have to dig into the implementation details. I don't like saying that these developers use Pulsar in a wrong way. I'd like to say they are limited by the poor interface that Pulsar provided. Keeping the limitation is harmful to the Pulsar community and could make ecosystem applications flaky and easy to break.

[BewareMyPower]

it takes a long time to correct it and I don't want to see it again.

The point is, now it's open sourced and many more external applications need such ability to access these internal fields. When Pulsar was not open sourced, the use cases were limited. But things are different now.

[BewareMyPower]

I don't have much time to keep debating on this. And it already spent me too much time to clarify the motivation and find the code examples.

Let's use [email protected] to make decisions and hear more voices.

[rdhabalia]

I don't think Pulsar has poor interface or has any limitation. MessageId is just a reference from Pulsar and it should have correct serialization and deserialization methods. I don't know any system which encourages or provide a contract to extract messageId and depend on internal.
So, I don't think it's fair to say that Pulsar has limitations and it has poor APIs or interfaces. Abstraction is an important part of API contract and that's what Pulsar is doing. why keeping MessageID abstract from user can break application? and that's the exact reason why user application should not depend on such abstraction and should not try to hack that abstraction.

[rdhabalia]

When Pulsar was not open sourced, the use cases were limited. But things are different now.

It's not about usecases but it's about learning to do the right things.

[BewareMyPower]

Pulsar Functions converts the MessageId to MessageIdImpl to get the ledger ID and entry ID:

pulsar/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java

Lines 328 to 330 in d1fc732

	MessageIdImpl msgId = MessageIdImpl.convertToMessageIdImpl(messageId);
	long ledgerId = msgId.getLedgerId();
	long entryId = msgId.getEntryId();

Kafka Connect Sink even uses reflection to get the batch index from MessageId:

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

Line 404 in d1fc732

try {

Pulsar Spark Connector casts MessageId to the implementations for these internal fields:

https://github.com/streamnative/pulsar-spark/blob/fa131552ab5e857f801b91a58f97b8a09bf3eecc/src/main/scala/org/apache/spark/sql/pulsar/PulsarSources.scala#L73

The Spark case is extremely terrible, because it depends on many more internal methods like TopicMessageIdImpl#getInnerMessageId.

  def mid2Impl(mid: MessageId): MessageIdImpl = {
    mid match {
      case bmid: BatchMessageIdImpl =>
        new MessageIdImpl(bmid.getLedgerId, bmid.getEntryId, bmid.getPartitionIndex)
      case midi: MessageIdImpl => midi
      case t: TopicMessageIdImpl => mid2Impl(t.getInnerMessageId)
      case up: UserProvidedMessageId => mid2Impl(up.mid)
    }
  }

Before considering what if we replace bookkeeper with other some other storage?, I'd like to ask:

• What if the getInnerMessageId method was removed?
• What if the TopicMessageIdImpl class was removed?

I believe can find more cases the external applications want to access these internal fields. Even if we didn't encourage them to do that. The MessageIdAdv interface is not added to encourage users to use internal data. Instead, many users already assume the storage is BK and want to use internal data but they have to access unstable methods in the implementations module (pulsar-client). It brings the problem that each minor modification of the implementation could bring breaking changes.

[Shawyeok]

Add MessageIdAdvUtils for the most common used methods.
Replace BatchMessageAcker with the BitSet for ACK.

There is a race condition on acknowledge with batch message, details #22352