apache / apache/pulsar

PIP-224: Introduce TopicMessageId for consumer's MessageId related APIs

Open
#18,616 8 comments 1 reaction 0 assignees View on GitHub
Stale type/PIP
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Motivation

There are two ways to get a `MessageId` from consumer:
- `Message#getMessageId` from a received message
- `Consumer#getLastMessageId`

The returned `MessageId` can be used in `seek` and `acknowledge` (including `acknowledgeCumulative`). A `MessageId` represents the position of a message. For a consumer that subscribes a single topic, things are usually right because the MessageId from a received message or returned by `getLastMessageId` always belongs to the topic. However, it gets different for a consumer that subscribes multiple topics (let's say **multi-topics consumer**).

For a multi-topics consumer:
- `getLastMessageId` returns a `MultiMessageIdImpl`. However, it's a meaningless implementation. It maintains a map that maps topic name to a `MessageId`. It's not comparable and serializable. And it's very weird to compare a MessageId and a `MultiMessageIdImpl` that represents multiple messages' positions.
- `seek` only accepts an earliest or latest position. Maybe it's because there is no way to know which topic the message belongs to.
- `acknowledge` only accepts a `TopicMessageIdImpl` implementation with a `getTopicPartitionName()` method to know which topic the message belongs to. However, when the topic is not subscribed by the consumer, the behavior is undefined. The current behaviors are:
- If the topic is not subscribed, no error is returned.
- If the `MessageId` is not a `TopicMessageIdImpl`, `IllegalArgumentException` will be thrown.

### Goal

This proposal will introduce a `TopicMessageId` interface that exposes a method to get a message's owner topic. Therefore, we can make these behaviors more clear with `TopicMessageId`.

### API Changes

```java
/**
* The MessageId used for a consumer that subscribes multiple topics or partitioned topics.
*
*


* It's guaranteed that {@link Message#getMessageId()} must return a TopicMessageId instance if the Message is received
* from a consumer that subscribes multiple topics or partitioned topics.
* The topic name used in APIs related to this class like `getOwnerTopic` and `create` must be the full topic name. For
* example, "my-topic" is invalid while "persistent://public/default/my-topic" is valid.
* If the topic is a partitioned topic, the topic name should be the name of the specific partition, e.g.
* "persistent://public/default/my-topic-partition-0".
*


*/
public interface TopicMessageId extends MessageId {

/**
* Return the owner topic name of a message.
*
* @return the owner topic
*/
String getOwnerTopic();

static TopicMessageId create(String topic, MessageId messageId) {
if (messageId instanceof TopicMessageId) {
return (TopicMessageId) messageId;
}
return new TopicMessageId() {

@Override
public String getOwnerTopic() {
return topic;
}

@Override
public byte[] toByteArray() {
return messageId.toByteArray();
}

@Override
public int compareTo(MessageId o) {
return messageId.compareTo(o);
}
};
}
}
```

The `create` method can be used to replace the `TopicMessageIdImpl` implementation. Once we have a `MessageId` returned by `Message#getMessageId()`, we can use `TopicMessageId.create(topic, msgId)` to have a message id to seek or acknowledge.

> **NOTE**
>
> This `create` method is actually not required for acknowledgment. Because the `MessageId` in the received message should be the correct implementation. So the previous code could work without explicitly using the new API.
>
> It's only used for new seek APIs because they accept a `TopicMessageId`.

Add a `TopicMessageIdSerDes` class to serialize and deserialize a `TopicMessageId`:

```java
/**
* To keep the backward compatibility, {@link TopicMessageId#toByteArray()} should not serialize the owner topic. This
* class provides a convenient way for users to serialize a TopicMessageId with its owner topic serialized.
*/
class TopicMessageIdSerDes {
public static byte[] serialize(TopicMessageId topicMessageId) {/* ... */}
public static TopicMessageId deserialize(byte[] bytes) {/* ... */}
}
```

Here we don't override the `toByteArray()` method for backward compatibility because there might be existing code like:

```java
var msg = multiTopicsConsumer.receive();
// msg.getMessageId() is a TopicMessageId
serialize(multiTopicsConsumer.getTopic(), msg.getMessageId().toByteArray());
```

This interface doesn't add any `acknowledge` overload because the overloads are already too many. But it will make the behavior clear.

```java
/**
* ...
* @throws PulsarClientException.NotAllowedException if `messageId` is not a {@link TopicMessageId} when multiple topics are subscribed.
*/
void acknowledge(MessageId messageId) throws PulsarClientException;

// NOTE: the same goes for acknowledgeCumulative

/**
* ...
* @throws PulsarClientException.NotAllowedException if any message id in the list is not a {@link TopicMessageId} when multiple topics are subscribed.
*/
void acknowledge(List messageIdList) throws PulsarClientException;
```

For `seek` operations, the semantics will be modified when the argument is a `TopicMessageId`:

```java
/**
* ...
*

Note: For multi-topics consumer, if `messageId` is a {@link TopicMessageId}, the seek operation will happen
* on the owner topic of the message, which is returned by {@link TopicMessageId#getOwnerTopic()}. Otherwise, you
* can only seek to the earliest or latest message.
* ...
*/
void seek(MessageId messageId) throws PulsarClientException;
```

i.e. for a multi-topics consumer, we can now seek the message id of a received message without any code change:

```java
var msgId = multiTopicsConsumer.receive().getMessageId(); // it's a TopicMessageId
consumer.seek(msgId); // seek(TopicMessageId) will be called, before this PIP an exception would be thrown
```

It's because the `MessageId` of a message received from a multi-topics consumer is a `TopicMessageId` that already contains the correct topic name.

For those want to convert a `MessageId`, which might not be a `TopicMessageId`, to a `TopicMessageId` if they know the topic name, they can perform the conversion explicitly like:

```java
consumer.seek(TopicMessageId.create(msg.getTopicName(), msgId);
```

It's used when users want to seek a `MessageId` returned by `Producer#send`. In this case, the `MessageId` is not a `TopicMessageId` and we have to pass the topic name to tell the multi-topics consumer which topic it belongs to.

For `getLastMessageAsync` method, change the semantics to disallow it on a multi-topics consumer. Instead, add these two APIs:

```java
// NOTE: it's guaranteed that no duplicated owner topics should be returned
List getLastMessageIds() throws PulsarClientException;
CompletableFuture> getLastMessageIdsAsync();
```

### Implementation

The main changes are removing the `TopicMessageIdImpl` and `MultiTopicMessageIdImpl`. Then add the ability to seek or acknowledge a specific topic with a `MessageId` on a multi-topics consumer. Much code and logic could be reused.

### Alternatives

https://lists.apache.org/thread/vydpjgyrfzkr45ho9r12sd2lw5c4mw6s tries to seek or acknowledge a topic and a `MessageId`. But it will increase many public APIs. With this proposal, new APIs are only added for the missed features (e.g. seek a `MessageId` other than `earliest` or `latest` on a multi-topic consumer).

Adding a `seek(String, MessageId)` overload is also denied for two reasons:
1. There is only one valid topic for a given `MessageId`. Users have to get the correct topic name, e.g. `my-topic-partition-2`. Otherwise, the seek command might be sent to a wrong topic and unexpected behavior might happen.
2. Users have to distinguish whether the consumer is a multi-topics consumer. If not, they need to use the original `seek(MessageId)` method. It's an inconsistency for user experience.

Adding a `serialize` method and a static `deserialize` method (like `toByteArray()` and `fromByteArray` methods of `MessageId`) to `TopicMessageId` is denied: the SerDes implementation of `TopericMessageId` should be unique. We should not allow other implementations.

Inheriting a `Schema` is also denied: there are some other methods in `Schema` like `getSchemaInfo` that might make users confused. We only need a clear SerDes implementation of `TopicMessageId`.

### Anything else?

See previous discussions related to the `MessageId`:
- [[DISCUSSION] Support seek and ack multiple MessageIds](https://lists.apache.org/thread/rdkqnkohbmkjjs61hvoqplhhngr0b0sd)
- [[DISCUSSION] Redesign the MessageId interface](https://lists.apache.org/thread/vydpjgyrfzkr45ho9r12sd2lw5c4mw6s)

Contributor guide

Open the contributing guide

Research direction

Start by reading the existing MessageId and Consumer APIs, especially TopicMessageIdImpl and MultiTopicMessageIdImpl, then review the proposed TopicMessageId and TopicMessageIdSerDes contracts. Done means the replacement API and serialization behavior are implemented, with multi-topic seek, acknowledge, and last-message operations following the stated semantics.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend-api-design, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.