[Enhancement] Improve the performance of publish event in connector-pulsar

[fengyongshe]

Search before asking

• I had searched in the issues and found no similar issues.

Enhancement Request

When pulsar connector receive a produce event , will construct a new producer instance. It's in a wrong way and poor performance .

public void publish(CloudEvent cloudEvent, SendCallback sendCallback) {

     try {
         Producer<byte[]> producer = this.pulsarClient.newProducer()
                 .topic(cloudEvent.getSubject())
                 .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                 .sendTimeout(10, TimeUnit.SECONDS)
                 .blockIfQueueFull(true)
                 .create();

         ......
         producer.sendAsync(serializedCloudEvent).thenAccept(messageId -> {
             sendCallback.onSuccess(CloudEventUtils.convertSendResult(cloudEvent));
         });
     }

As the code above shows , One message one Producer .

Describe the solution you'd like

1. Construct the producer in init , and put it in the Cache(HashMap)
2. When publish a event, get it from cache
3. If need to create a new instance, do as the step one

Are you willing to submit PR?

• Yes I am willing to submit a PR!

[fengyongshe]

@xwm1992 I'm willing to accomplish it .