[fix][client] Use dedicated executor for requests in BinaryProtoLookupService

[nodece]

Motivation

I'm testing Pulsar 3.0.6. We have a Pulsar interceptor, which records the data(web request/binary command) to the topic by the producer, but I got this error:

[ZKMetadataStore-12-1] INFO org.apache.pulsar.broker.service.ServerCnx - [/192.168.100.102:60796] Created subscription on topic persistent://as/system/__change_events / reader-e710cb132a
[pulsar-io-6-2] WARN org.apache.pulsar.client.impl.ClientCnx - [/192.168.100.102:15002] Got exception java.lang.UnsupportedOperationException
	at org.apache.pulsar.common.protocol.PulsarDecoder.handlePartitionMetadataRequest(PulsarDecoder.java:508)
	at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:134)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.handler.flush.FlushConsolidationHandler.channelRead(FlushConsolidationHandler.java:152)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)

This indicates the broker sends a PartitionMetadataRequest command to the client, this is a bizarre behavior.

How to reproduce this issue?

1. Create a producer asynchronously in the org.apache.pulsar.broker.intercept.BrokerInterceptor#onPulsarCommand.
2. The producer will change the type of pulsar response command, which will be replaced with PARTITION_METADATA.

Due to Pulsar's use of ThreadLocal for command creation, the command instances are being unexpectedly modified.

Modifications

• Add lookup executor for requests in BinaryProtoLookupService.
• Add shared lookup executor for the broker client.

Documentation

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

[lhotari]

Due to Pulsar's use of ThreadLocal for command creation, the command instances are being unexpectedly modified.

Thanks for bringing this up. This seems to be a severe bug and it would justify to report it separately. It's easy to miss this PR that this is addressing a critical issue.

[lhotari]

Create a producer asynchronously in the org.apache.pulsar.broker.intercept.BrokerInterceptor#onPulsarCommand.

Is this the only case where this can be reproduced?

[lhotari]

1. Create a producer asynchronously in the org.apache.pulsar.broker.intercept.BrokerInterceptor#onPulsarCommand.

@nodece I think that the correct solution would be to call new BaseCommand().copyFrom(command) in the interceptor to make a copy of the command. That's what is needed whenever the thread locals are referenced later.

[lhotari]

The correct solution for keeping a reference in the onPulsarCommand broker interceptor implementation would be to use new BaseCommand().copyFrom(command). That's why I don't think that the solution in the PR directly related to addressing the problem.

[nodece]

Create a producer asynchronously in the org.apache.pulsar.broker.intercept.BrokerInterceptor#onPulsarCommand.

Is this the only case where this can be reproduced?

At present, it looks like this.

@nodece I think that the correct solution would be to call new BaseCommand().copyFrom(command) in the interceptor to make a copy of the command. That's what is needed whenever the thread locals are referenced later.

My interceptor doesn't change the command. Please notice the 1 and 2 in my picture, the broker and client in the same k thread.

1. org.apache.pulsar.broker.service.PulsarCommandSenderImpl#sendSuccessResponse:

    public void sendSuccessResponse(long requestId) {
        // step-1: caller in the zk thread.
        BaseCommand command = Commands.newSuccessCommand(requestId);
        // step-2: call the interceptor method.
        safeIntercept(command, cnx);
        ByteBuf outBuf = Commands.serializeWithSize(command);
        writeAndFlush(outBuf);
    }

Use the ThreadLocal to create the command in the zk thread.

2. My intercept method:

    public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
        // caller in the zk thread.
        Producer<byte[]> producer = client.newProducer().topic("my-topic").create();
        producer.send(command.getType().name().getBytes(StandardCharsets.UTF_8));
    }

The produce creation needs to do the lookup and then get partition metadata, which requests the broker by Commands.newLookup and Commands.newPartitionMetadataRequest commands.

We are using the ThreadLocal to create the lookup or partition metadata requests, so the sendSuccessResponse command is broken in the step-2.

The current solution version uses a dedicated executor to create commands and make requests in the client.

Pulsar uses implicit executor in many places, so often we are not sure who the executor is, as a result, this accident occurred, clarifying the executor is not a bad thing.

[lhotari]

Pulsar uses implicit executor in many places, so often we are not sure who the executor is, as a result, this accident occurred, clarifying the executor is not a bad thing.

It's very hard to be sure whether it had performance impacts. When an executor is defined, it will add extra overhead of queuing to the other executor and the extra thread switching overhead. It depends a lot on the situation whether this causes performance regressions or not.

[nodece]

Do you have any suggestions or recommendations for resolving this issue?

[lhotari]

Do you have any suggestions or recommendations for resolving this issue?

@nodece The changes in this PR LGTM. There's no risk of additional overhead of using the separate executor since the number of ops is relatively low for the operations that it applies to. I'll approve this PR. Thanks for the contribution!

[lhotari]

LGTM

A lot of tests fail. This is causing problems.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.06977% with 9 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.32%. Comparing base (bbc6224) to head (4c3db76).
⚠️ Report is 1291 commits behind head on master.

Files with missing lines	Patch %	Lines
...e/pulsar/client/impl/BinaryProtoLookupService.java	80.95%	3 Missing and 1 partial ⚠️
...rg/apache/pulsar/client/impl/PulsarClientImpl.java	77.77%	4 Missing ⚠️
...n/java/org/apache/pulsar/broker/PulsarService.java	75.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23378      +/-   ##
============================================
+ Coverage     73.57%   74.32%   +0.75%     
- Complexity    32624    34402    +1778     
============================================
  Files          1877     1943      +66     
  Lines        139502   146973    +7471     
  Branches      15299    16191     +892     
============================================
+ Hits         102638   109239    +6601     
- Misses        28908    29302     +394     
- Partials       7956     8432     +476     

Flag	Coverage Δ
inttests	27.65% <58.13%> (+3.06%)	⬆️
systests	24.40% <51.16%> (+0.08%)	⬆️
unittests	73.70% <79.06%> (+0.85%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...n/java/org/apache/pulsar/broker/PulsarService.java	84.28% <75.00%> (+1.91%)	⬆️
...e/pulsar/client/impl/BinaryProtoLookupService.java	81.16% <80.95%> (-1.38%)	⬇️
...rg/apache/pulsar/client/impl/PulsarClientImpl.java	75.20% <77.77%> (+0.90%)	⬆️

... and 647 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[nodece]

Ping @lhotari

[dao-jun]

lgtm

[lhotari]

LGTM, good work @nodece

[nodece]

It was cherry-picked to the branch-3.0 by #23461.

[lhotari]

It was cherry-picked to the branch-3.0 by #23461.

@nodece Please notify the dev mailing list about this change. This PR could potentially cause regressions so it's better to keep others informed about the change in branch-3.0. I'm not against changing this, but it's hard to know if there are regressions before it has been extensively tested. Are you going to test branch-3.0 with this change?

[nodece]

@lhotari Thanks for your reminder! I sent a notice to the dev mailing list.

Are you going to test branch-3.0 with this change?

I tested this change in a private ecosystem, it works fine.