[feat][pcip] PCIP-5 Support Pull Consumer

[StevenLuMT]

PCIP Author

• @liangyepianzhou
• @StevenLuMT

Motivation

Our organization is currently planning a migration from RocketMQ to Pulsar. To facilitate this transition, we aim to implement a standardized abstraction layer for MQ clients that encapsulates implementation details of specific messaging systems. This abstraction layer will allow seamless engine replacement while maintaining consistent client interfaces. However, two critical compatibility issues hinder the unification of message fetching patterns between RocketMQ/Kafka and Pulsar:

1. Interface Disparity:

• Pulsar currently lacks native support for offset-based message fetching (fetch/pull paradigm) that allows specifying exact message positions and batch sizes.

2. Positioning Mechanism Mismatch:

• RocketMQ/Kafka: Utilize monotonically increasing numerical offsets for message positioning and acknowledgment
• Pulsar: Relies on composite MessageID (ledgerId + entryId + batchIndex) for message identification.

Therefore, I propose to contribute a customized feature implementation to the Pulsar contributor repository that leverages existing Pulsar APIs to achieve functionality similar to RocketMQ's pull consumer. This implementation could serve companies like ours with standardized client requirements, allowing them to reuse this solution.

Modifications

1. Upgrade Pulsar version to 4.0.8(features include PIP-415 a new API to get the message id by offset) Upgrade Pulsar version to 4.1.1 #23
2. Implement RocketMQ-like pull consumer functionality using existing Pulsar APIs.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure