[improve][pcip] PCIP-2 Distributed RPC framework implemented by the Pulsar client #6

Denovo1998 · 2024-09-02T15:27:33Z

Feedback is not positive due to PIP-371.
apache/pulsar#23143
apache/pulsar#23194

We need to implement this distributed RPC framework in a way that does not intrude into the pulsar core library. Therefore, we need to use two topics, one is the request topic and the other is the reply topic. The client side sends RPC requests to the request topic, the server side receives request message and performs customized processing, and finally sends them to the reply topic. The client receives the reply message and returns.

Motivation

As we known，Pulsar's current asynchronous publish-subscribe model serves well for decoupled message distribution, but it lacks a native mechanism for handling synchronous interactions typical of Remote Procedure Calls (RPC).

This request-reply model can greatly enhance the utility of Pulsar. We can then use Pulsar as RPC.

Why would we use Pulsar for this RPC call?

Implement RPC using Apache Pulsar. Requests can be sent through a client, received by one or more servers and processed in parallel. Finally, the server returns all processing results after processing, and the client can perform summary and other operations after receiving them.
This proposal to achieve the function is request. Request and existing send function of pulsar can be mixed to same topic. This means that the user can choose, and the call to the server side (consumer) can be asynchronous or synchronous, which is controlled by the user flexibly.
You can directly use Pulsar's own delaying messages, that is, you can execute RPC regularly.
You can directly use Pulsar's own load balancing mechanism.
You can directly use Pulsar's own message consumption throttling mechanism.
You can directly use Pulsar's own expansion and contraction mechanism.
You can directly use Pulsar's own message call tracking, monitoring, and logging mechanisms.

Modifications

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

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/package-info.java

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/common/package-info.java

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/server/package-info.java

pulsar-rpc-contrib/pom.xml

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/ReplyListener.java

StevenLuMT · 2024-09-03T01:43:35Z

this image has some Chinese,please translate to English

pulsar-rpc-contrib/src/test/java/org/apache/pulsar/rpc/contrib/SimpleRpcCallTest.java

…e version definitions to parent

Denovo1998 · 2024-09-03T04:46:23Z

this image has some Chinese,please translate to English

Picture updated.

AuroraTwinkle · 2024-09-09T02:33:28Z

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/PulsarRpcClient.java

+ * @param <V> The type of the reply messages.
+ */
+@RequiredArgsConstructor(access = PACKAGE)
+public class PulsarRpcClient<T, V> implements AutoCloseable {


It is recommended to expose the interface externally, refer to PulsarClient.

Maybe same as #6 (comment)

PulsarRpcClient, PulsarRpcClientImpl, PulsarRpcClientBuilder, PulsarRpcClientBuilderImpl
Already added.

PTAL!

If this method is correct, I will modify the server code later.

AuroraTwinkle · 2024-09-09T02:39:24Z

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/server/RequestListener.java

+ */
+@Slf4j
+@RequiredArgsConstructor
+public class RequestListener<T, V> implements MessageListener<T> {


If you do not need to expose this class to the business, you can use package-level private.

AuroraTwinkle · 2024-09-09T02:39:35Z

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/ReplyListener.java

+ */
+@Slf4j
+@RequiredArgsConstructor
+public class ReplyListener<V> implements MessageListener<V> {


If you do not need to expose this class to the business, you can use package-level private.

liangyepianzhou · 2024-09-09T02:43:24Z

Feedback is not positive due to PIP-371. apache/pulsar#23143 apache/pulsar#23194

We need to implement this distributed RPC framework in a way that does not intrude into the pulsar core library. Therefore, we need to use two topics, one is the request topic and the other is the reply topic. The client side sends RPC requests to the request topic, the server side receives request message and performs customized processing, and finally sends them to the reply topic. The client receives the reply message and returns.

Motivation

As we known，Pulsar's current asynchronous publish-subscribe model serves well for decoupled message distribution, but it lacks a native mechanism for handling synchronous interactions typical of Remote Procedure Calls (RPC).

This request-reply model can greatly enhance the utility of Pulsar. We can then use Pulsar as RPC.

Why would we use Pulsar for this RPC call?

This proposal to achieve the function is request. Request and existing send function of pulsar can be mixed to same topic. This means that the user can choose, and the call to the server side (consumer) can be asynchronous or synchronous, which is controlled by the user flexibly.

You can directly use Pulsar's own delaying messages, that is, you can execute RPC regularly.

You can directly use Pulsar's own load balancing mechanism.

You can directly use Pulsar's own message consumption throttling mechanism.

You can directly use Pulsar's own expansion and contraction mechanism.

You can directly use Pulsar's own message call tracking, monitoring, and logging mechanisms.

Another benefit of using message queues to implement RPC requests is that the request message can be received by multiple services and processed in parallel, and then all the processing results are returned. Finally, the client will do the integration.

liangyepianzhou

Good work! I left some comments. The most important point is the need for a concise API. Make it simple enough for users to use.
A simplest model:

newClient()
reply send Request (k, v)

pom.xml

liangyepianzhou · 2024-09-09T02:53:29Z

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/PulsarRpcClient.java

+     * @return The reply value.
+     * @throws Exception if an error occurs during the request or while waiting for the reply.
+     */
+    public V request(String correlationId, TypedMessageBuilder<T> message) throws Exception {


The interface should return a specific exception. Here you can catch the interrupt exception and throw a custom exception.

We'd better not expose the MQ implementation. Only key and value are passed in here.
This will make it easier for users to use.

We'd better not expose the MQ implementation. Only key and value are passed in here. This will make it easier for users to use.

It is also considered here that some custom configurations may be made for the property or eventTime of the message. It is a bit too inflexible to pass only key and value. It may be better for users to construct a message passed in by themselves.

It allows users to pass in config. As I said before, you can load more configuration files for Producer and Consumer through .loadConf(). You can also pass in message configuration through message.loadConf().

Each message and the request producer has been modified using the loadConf().

...r-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/DefaultRequestCallBack.java

liangyepianzhou · 2024-09-09T03:33:47Z

...rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/common/MessageDispatcherFactory.java

+    public Producer<T> requestProducer(ProducerBuilder<T> requestProducer) throws IOException {
+        return requestProducer
+                // allow only one client
+                .accessMode(ProducerAccessMode.Exclusive)


We can expose a way to modify the Producer and consumer configuration by .loadConf().

Each message and the request producer has been modified using the loadConf().

liangyepianzhou · 2024-09-09T03:41:37Z

pulsar-rpc-contrib/src/main/java/org/apache/pulsar/rpc/contrib/client/PulsarRpcClient.java

+     * @return A new instance of {@link PulsarRpcClient}.
+     * @throws IOException if there is an error during the client initialization.
+     */
+    public static <T, V> PulsarRpcClient<T, V> create(


I think we need an implementation that doesn't expose the internal MQ client.

Maybe same as #6 (comment)

PulsarRpcClient, PulsarRpcClientImpl, PulsarRpcClientBuilder, PulsarRpcClientBuilderImpl
Already added.

PTAL!

If this method is correct, I will modify the server code later.

Denovo1998 · 2024-09-09T13:45:04Z

Feedback is not positive due to PIP-371. apache/pulsar#23143 apache/pulsar#23194
We need to implement this distributed RPC framework in a way that does not intrude into the pulsar core library. Therefore, we need to use two topics, one is the request topic and the other is the reply topic. The client side sends RPC requests to the request topic, the server side receives request message and performs customized processing, and finally sends them to the reply topic. The client receives the reply message and returns.

Motivation

As we known，Pulsar's current asynchronous publish-subscribe model serves well for decoupled message distribution, but it lacks a native mechanism for handling synchronous interactions typical of Remote Procedure Calls (RPC).
This request-reply model can greatly enhance the utility of Pulsar. We can then use Pulsar as RPC.
Why would we use Pulsar for this RPC call?

This proposal to achieve the function is request. Request and existing send function of pulsar can be mixed to same topic. This means that the user can choose, and the call to the server side (consumer) can be asynchronous or synchronous, which is controlled by the user flexibly.

You can directly use Pulsar's own delaying messages, that is, you can execute RPC regularly.

You can directly use Pulsar's own load balancing mechanism.

You can directly use Pulsar's own message consumption throttling mechanism.

You can directly use Pulsar's own expansion and contraction mechanism.

You can directly use Pulsar's own message call tracking, monitoring, and logging mechanisms.

Another benefit of using message queues to implement RPC requests is that the request message can be received by multiple services and processed in parallel, and then all the processing results are returned. Finally, the client will do the integration.

Has been updated.

StevenLuMT · 2024-09-15T23:07:00Z

@liangyepianzhou please agree to the workflow and do some checks first.

Denovo1998 · 2024-09-16T14:52:22Z

@liangyepianzhou @AuroraTwinkle @StevenLuMT
Currently only SimpleRpcCallTest#testRpcCallWithCallBack has some problems. Can you help me look at it? It may be related to topicsPattern.

Denovo1998 · 2024-09-29T13:54:18Z

@StevenLuMT I added the testcontainers, and fixed the SimpleRpcCallTest#testRpcCallWithCallBack. Please agree to the workflow.

pulsar-rpc-contrib/src/test/java/org/apache/pulsar/rpc/contrib/PulsarRpcBase.java

StevenLuMT · 2024-10-02T22:43:00Z

about Question5
I will confirm this question with you again. It may also be related to translation. My suggestion logic at that time was:

Can you add a few confirmation methods, such as writing needs to wait for the following strategies to return success, similar to these scenarios:

Scenario 1: All subscriptions to sub1/sub2 are acked successfully
Scenario 2: n (0<n<All) subscriptions to sub1/sub2 are acked successfully
Scenario 3: No need to wait for the ack of subscriptions to sub1/sub2, fall back to the original mq mode

Denovo1998 · 2024-10-03T04:33:17Z

about Question5 I will confirm this question with you again. It may also be related to translation. My suggestion logic at that time was:

Can you add a few confirmation methods, such as writing needs to wait for the following strategies to return success, similar to these scenarios:

Scenario 1: All subscriptions to sub1/sub2 are acked successfully Scenario 2: n (0<n<All) subscriptions to sub1/sub2 are acked successfully Scenario 3: No need to wait for the ack of subscriptions to sub1/sub2, fall back to the original mq mode

Users can implement a org.apache.pulsar.rpc.contrib.client.RequestCallBack and implement onReplySuccess methods to achieve the three scenarios described above.

/**
 * Invoked after receiving a reply from the server successfully.
 *
 * <p>Please note that {@code replyFuture.complete(value)} must be executed at the end.
 *
 * @param correlationId The correlation ID associated with the reply.
 * @param subscription The subscription name the reply was received on.
 * @param value The value of the reply.
 * @param replyFuture The future to be completed with the received value.
 */
void onReplySuccess(String correlationId, String subscription, V value, CompletableFuture<V> replyFuture);

When the ReplyListener receives the reply message, it directly calls onReplySuccess, and the parameter contains the subscription, key, and value to which it belongs. If the user-defined "success" criteria are met, the user will call rpcClient.removeRequest(correlationId); remove this request. At this time, if the ReplyListener still receives reply message corresponding to non-existent request, it will directly ack and not return to the user.

So the framework itself does not provide the corresponding scenario implementation, but through the RequestCallBack, users can achieve a variety of scenarios.

I added a simple scenario implementation in SimpleRpcCallTest#testRpcCallWithCallBack.
@StevenLuMT

StevenLuMT

Currently, a module pulsar-rpc-contrib has been added, and a PIP needs to be proposed. I think we should take this opportunity to vote and decide how to merge the current project in the future, similar to what lari proposed, https://lists.apache.org/thread/y2ccnoxj76pnw25n5p91ymqygjoq1x27

My suggestion is to be consistent with bookkeeper's method. I will initiate a vote:

liangyepianzhou · 2024-10-09T02:51:36Z

Currently, a module pulsar-rpc-contrib has been added, and a PIP needs to be proposed. I think we should take this opportunity to vote and decide how to merge the current project in the future, similar to what lari proposed, https://lists.apache.org/thread/y2ccnoxj76pnw25n5p91ymqygjoq1x27

My suggestion is to be consistent with bookkeeper's method. I will initiate a vote:

Because this project belongs to Pulsar, I think it is better to initiate discussions and votes on Pulsar's mailing list.
And Pulsar is also optimizing its voting process to reduce blocks.
https://lists.apache.org/thread/5lv8pwdgtxcobkjfgto274xxvrr3b1q7

Denovo1998 · 2024-11-19T13:15:05Z

@StevenLuMT @liangyepianzhou @AuroraTwinkle The PCIP document has been updated. PTAL!

StevenLuMT

Good Job

pcip/pcip-2.md

liangyepianzhou · 2024-12-01T14:56:59Z

@lhotari @eolivelli Would you be willing to help review this PR when you have time?

Denovo1998 added 4 commits August 29, 2024 23:23

RPC framework implemented by the Pulsar client

caa6e8e

rpc server, timeout test, error on server side test, pattern test

4a5f022

support server-side multi-subscription, request callback on client side

5c4add2

testRpcCallWithCallBack, removeRequest

5e34c80