Add missed serialization schema argument for PulsarOutputFormat const…

…ructor (apache#4373) The constructor of PulsarOutputFormat expects a serializationSchema passed in otherwise it's member serializationSchema will be assigned by itself.
vruc · May 29, 2019 · 4d1b2ff · 4d1b2ff
1 parent ba24d73
commit 4d1b2ff
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 5 deletions.
diff --git a/pulsar-flink/src/main/java/org/apache/flink/batch/connectors/pulsar/PulsarOutputFormat.java b/pulsar-flink/src/main/java/org/apache/flink/batch/connectors/pulsar/PulsarOutputFormat.java
@@ -37,7 +37,9 @@ public PulsarOutputFormat(String serviceUrl, String topicName, Authentication au
         this.serializationSchema = serializationSchema;
     }
 
-    public PulsarOutputFormat(ClientConfigurationData clientConfigurationData, ProducerConfigurationData producerConfigurationData) {
+    public PulsarOutputFormat(final ClientConfigurationData clientConfigurationData,
+                              final ProducerConfigurationData producerConfigurationData,
+                              final SerializationSchema<T> serializationSchema) {
         super(clientConfigurationData, producerConfigurationData);
         Preconditions.checkNotNull(serializationSchema, "serializationSchema cannot be null.");
         this.serializationSchema = serializationSchema;

diff --git a/...-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/PulsarOutputFormatTest.java b/...-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/PulsarOutputFormatTest.java
@@ -72,7 +72,7 @@ public void testPulsarOutputFormatConstructorV2WhenServiceUrlIsNull() {
                 .topicName("testTopic")
                 .build();
 
-        new PulsarOutputFormat(clientConf, producerConf);
+        new PulsarOutputFormat(clientConf, producerConf, text -> text.toString().getBytes());
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
@@ -85,7 +85,7 @@ public void testPulsarOutputFormatConstructorV2WhenTopicNameIsNull() {
                 .topicName(null)
                 .build();
 
-        new PulsarOutputFormat(clientConf, producerConf);
+        new PulsarOutputFormat(clientConf, producerConf, text -> text.toString().getBytes());
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
@@ -98,7 +98,7 @@ public void testPulsarOutputFormatConstructorV2WhenTopicNameIsBlank() {
                 .topicName(StringUtils.EMPTY)
                 .build();
 
-        new PulsarOutputFormat(clientConf, producerConf);
+        new PulsarOutputFormat(clientConf, producerConf, text -> text.toString().getBytes());
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
@@ -111,7 +111,18 @@ public void testPulsarOutputFormatConstructorV2WhenServiceUrlIsBlank() {
                 .topicName("testTopic")
                 .build();
 
-        new PulsarOutputFormat(clientConf, producerConf);
+        new PulsarOutputFormat(clientConf, producerConf, text -> text.toString().getBytes());
+    }
+
+    @Test(expectedExceptions = NullPointerException.class)
+    public void testPulsarOutputFormatConstructorV2WhenSerializationSchemaIsNull() {
+        ClientConfigurationData clientConf = ClientConfigurationData.builder()
+                .serviceUrl("testServiceUrl")
+                .build();
+        ProducerConfigurationData producerConf = ProducerConfigurationData.builder()
+                .topicName("testTopic")
+                .build();
+        new PulsarOutputFormat(clientConf, producerConf, null);
     }
 
     @Test