add Schema cache

streamnative · Mar 18, 2021 · 9169ef5 · 9169ef5
1 parent 97604fd
commit 9169ef5
Showing 1 changed file with 14 additions and 1 deletion.
diff --git a/.../org/apache/flink/streaming/connectors/pulsar/table/DynamicPulsarSerializationSchema.java b/.../org/apache/flink/streaming/connectors/pulsar/table/DynamicPulsarSerializationSchema.java
@@ -81,6 +81,8 @@ class DynamicPulsarSerializationSchema
 
     private String valueFormatType;
 
+    private volatile Schema<RowData> schema;
+
     DynamicPulsarSerializationSchema(
             @Nullable SerializationSchema<RowData> keySerialization,
             SerializationSchema<RowData> valueSerialization,
@@ -193,6 +195,17 @@ public void setNumParallelInstances(int numParallelInstances) {
 
     @Override
     public Schema<RowData> getSchema() {
+        if (schema == null) {
+            synchronized (this) {
+                if (schema == null) {
+                    schema = buildSchema();
+                }
+            }
+        }
+        return schema;
+    }
+
+    private FlinkSchema<RowData> buildSchema() {
         if (StringUtils.isBlank(valueFormatType)) {
             return new FlinkSchema<>(Schema.BYTES.getSchemaInfo(), valueSerialization, null);
         }
@@ -206,7 +219,7 @@ private void hackPbSerializationSchema(Map<String, String> options) {
         // reflect read PbRowSerializationSchema#messageClassName
         if (valueSerialization instanceof PbRowDataSerializationSchema) {
             try {
-                String messageClassName =
+                final String messageClassName =
                         (String) FieldUtils.readDeclaredField(valueSerialization, "messageClassName", true);
                 options.put(PbFormatOptions.MESSAGE_CLASS_NAME.key(), messageClassName);
             } catch (IllegalAccessException e) {