[BUG] - Pulsar Producer leak

[david-streamlio]

Describe the bug
When using the connector for a single topic, a customer is seeing the number of connections to the broker increasing dramatically. After reviewing the heap dump, we have determined that the root cause is an excessive number of producer objects are getting created that are never closed.

Expected behavior
Only a single producer for a topic should be created and re-used.

Additional context
The issue was discussed on Slack, with the conclusion being that there is an error in the PulsarWriteTask.scala code that creates a new producer for each task instead of placing it in the cache and reusing it on subsequent calls.