[fix][client] SchemaDefinition handle JSR310_CONVERSION_ENABLED prope…

…rty (#20201) Signed-off-by: tison <wander4096@gmail.com>
apache · May 1, 2023 · a22700d · a22700d
1 parent e8d6395
commit a22700d
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 5 deletions.
diff --git a/...lient/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionBuilderImpl.java b/...lient/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionBuilderImpl.java
@@ -129,7 +129,7 @@ public SchemaDefinitionBuilder<T> withProperties(Map<String, String> properties)
         if (properties.containsKey(ALWAYS_ALLOW_NULL)) {
             alwaysAllowNull = Boolean.parseBoolean(properties.get(ALWAYS_ALLOW_NULL));
         }
-        if (properties.containsKey(ALWAYS_ALLOW_NULL)) {
+        if (properties.containsKey(JSR310_CONVERSION_ENABLED)) {
             jsr310ConversionEnabled = Boolean.parseBoolean(properties.get(JSR310_CONVERSION_ENABLED));
         }
         return this;

diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java
@@ -69,7 +69,7 @@ public void testAllowNullAvroSchemaCreate() {
     }
 
     @Test
-    public void testFillParametersToSchemainfo() {
+    public void testFillParametersToSchemaInfo() {
         Map<String, String> keyProperties = new TreeMap<>();
         keyProperties.put("foo.key1", "value");
         keyProperties.put("foo.key2", "value");
@@ -89,7 +89,6 @@ public void testFillParametersToSchemainfo() {
                         .build());
 
         Schema<KeyValue<Foo, Bar>> keyValueSchema1 = Schema.KeyValue(fooSchema, barSchema);
-
         assertEquals(keyValueSchema1.getSchemaInfo().getProperties().get("key.schema.type"), String.valueOf(SchemaType.AVRO));
         assertEquals(keyValueSchema1.getSchemaInfo().getProperties().get("key.schema.properties"),
                 "{\"__alwaysAllowNull\":\"true\",\"__jsr310ConversionEnabled\":\"false\",\"foo.key1\":\"value\",\"foo.key2\":\"value\"}");
@@ -98,6 +97,37 @@ public void testFillParametersToSchemainfo() {
                 "{\"__alwaysAllowNull\":\"true\",\"__jsr310ConversionEnabled\":\"false\",\"bar.key\":\"key\"}");
     }
 
+    @Test
+    public void testOverwriteSchemaDefaultProperties() {
+        Map<String, String> keyProperties = new TreeMap<>();
+        keyProperties.put("foo.key1", "value");
+        keyProperties.put("foo.key2", "value");
+        keyProperties.put(SchemaDefinitionBuilderImpl.ALWAYS_ALLOW_NULL, "false");
+        keyProperties.put(SchemaDefinitionBuilderImpl.JSR310_CONVERSION_ENABLED, "true");
+
+        Map<String, String> valueProperties = new TreeMap<>();
+        valueProperties.put("bar.key", "key");
+
+        AvroSchema<Foo> fooSchema = AvroSchema.of(
+                SchemaDefinition.<Foo>builder()
+                        .withPojo(Foo.class)
+                        .withProperties(keyProperties)
+                        .build());
+        AvroSchema<Bar> barSchema = AvroSchema.of(
+                SchemaDefinition.<Bar>builder()
+                        .withPojo(Bar.class)
+                        .withProperties(valueProperties)
+                        .build());
+
+        Schema<KeyValue<Foo, Bar>> keyValueSchema1 = Schema.KeyValue(fooSchema, barSchema);
+        assertEquals(keyValueSchema1.getSchemaInfo().getProperties().get("key.schema.type"), String.valueOf(SchemaType.AVRO));
+        assertEquals(keyValueSchema1.getSchemaInfo().getProperties().get("key.schema.properties"),
+                "{\"__alwaysAllowNull\":\"false\",\"__jsr310ConversionEnabled\":\"true\",\"foo.key1\":\"value\",\"foo.key2\":\"value\"}");
+        assertEquals(keyValueSchema1.getSchemaInfo().getProperties().get("value.schema.type"), String.valueOf(SchemaType.AVRO));
+        assertEquals(keyValueSchema1.getSchemaInfo().getProperties().get("value.schema.properties"),
+                "{\"__alwaysAllowNull\":\"true\",\"__jsr310ConversionEnabled\":\"false\",\"bar.key\":\"key\"}");
+    }
+
     @Test
     public void testNotAllowNullAvroSchemaCreate() {
         AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).withAlwaysAllowNull(false).build());

diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
@@ -44,8 +44,6 @@
 @Slf4j
 public class TopicSchema {
 
-    public static final String JSR_310_CONVERSION_ENABLED = "jsr310ConversionEnabled";
-    public static final String ALWAYS_ALLOW_NULL = "alwaysAllowNull";
     private final Map<String, Schema<?>> cachedSchemas = new HashMap<>();
     private final PulsarClient client;