[GitHub] [flink] tisonkun commented on a diff in pull request #20698: [FLINK-28609][Connector/Pulsar] PulsarSchema didn't get properly serialized

[GitBox <gi...@apache.org>]

tisonkun commented on code in PR #20698:
URL: https://github.com/apache/flink/pull/20698#discussion_r957200760


##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchema.java:
##########
@@ -118,67 +117,37 @@ public <K, V> PulsarSchema(
         SchemaInfo encodedInfo =
                 encodeKeyValueSchemaInfo(info.getName(), infoKey, infoValue, encodingType);
 
-        this.schemaInfo = encodeClassInfo(encodedInfo, KeyValue.class);
-        this.schema = createSchema(this.schemaInfo);
+        setSchemaInfo(encodeClassInfo(encodedInfo, KeyValue.class));
+    }
+
+    /** Validate the schema for having the required class info. */
+    private void setSchemaInfo(SchemaInfo schemaInfo) {
+        this.schema = createSchema(schemaInfo);
+        this.schemaInfo = schemaInfo;
+
+        this.schemaName = schemaInfo.getName();
+        this.schemaBytes = schemaInfo.getSchema();
+        this.schemaType = schemaInfo.getType();
+        this.schemaProperties = schemaInfo.getProperties();
     }
 
     public Schema<T> getPulsarSchema() {
+        if (schema == null) {
+            this.schema = createSchema(getSchemaInfo());
+        }
         return schema;
     }
 
     public SchemaInfo getSchemaInfo() {
+        if (schemaInfo == null) {
+            this.schemaInfo =
+                    new SchemaInfoImpl(schemaName, schemaBytes, schemaType, schemaProperties);
+        }
         return schemaInfo;
     }
 
     public Class<T> getRecordClass() {
-        return decodeClassInfo(schemaInfo);
-    }
-
-    private void writeObject(ObjectOutputStream oos) throws IOException {
-        // Name
-        oos.writeUTF(schemaInfo.getName());
-
-        // Schema
-        byte[] schemaBytes = schemaInfo.getSchema();
-        oos.writeInt(schemaBytes.length);
-        oos.write(schemaBytes);
-
-        // Type
-        SchemaType type = schemaInfo.getType();
-        oos.writeInt(type.getValue());
-
-        // Properties
-        Map<String, String> properties = schemaInfo.getProperties();
-        oos.writeInt(properties.size());
-        for (Map.Entry<String, String> entry : properties.entrySet()) {
-            oos.writeUTF(entry.getKey());
-            oos.writeUTF(entry.getValue());
-        }
-    }
-
-    private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IOException {
-        // Name
-        String name = ois.readUTF();
-
-        // Schema
-        int byteLen = ois.readInt();
-        byte[] schemaBytes = new byte[byteLen];
-        int read = ois.read(schemaBytes);

Review Comment:
   I suggest you simply replace this call with:
   
   ```java
   int read = IOUtils.tryReadFully(ois, schemaBytes);
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org