[flink] Allow to specify a custom Pulsar producer (apache#3894)

This is necessary in pretty much any non-trivial use-case. The ability
to control the settings of the Pulsar producer is paramount to
building real-life applications

pulsar-flink/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarProducer.java

@@ -115,13 +115,22 @@ public FlinkPulsarProducer(String serviceUrl,
                                String defaultTopicName,
                                SerializationSchema<IN> serializationSchema,
                                PulsarKeyExtractor<IN> keyExtractor) {
+        this(serviceUrl, defaultTopicName, serializationSchema, keyExtractor, null);
+    }
+
+    public FlinkPulsarProducer(String serviceUrl,
+                               String defaultTopicName,
+                               SerializationSchema<IN> serializationSchema,
+                               PulsarKeyExtractor<IN> keyExtractor,
+                               Producer<byte[]> producer) {
         checkArgument(StringUtils.isNotBlank(serviceUrl), "Service url cannot be blank");
         checkArgument(StringUtils.isNotBlank(defaultTopicName), "TopicName cannot be blank");
         this.serviceUrl = serviceUrl;
         this.defaultTopicName = defaultTopicName;
         this.schema = checkNotNull(serializationSchema, "Serialization Schema not set");
         this.flinkPulsarKeyExtractor = getOrNullKeyExtractor(keyExtractor);
         ClosureCleaner.ensureSerializable(serializationSchema);
+        this.producer = producer;
     }
 
     // ---------------------------------- Properties --------------------------
@@ -185,7 +194,10 @@ private Producer<byte[]> createProducer() throws Exception {
      */
     @Override
     public void open(Configuration parameters) throws Exception {
-        this.producer = createProducer();
+        if (producer == null) {
+            // If no custom producer was specified create a default one
+            this.producer = createProducer();
+        }
 
         RuntimeContext ctx = getRuntimeContext();