fix: make sure pulsar schema could serialize and deserialize in 2.10.2

flink-connectors/flink-connector-pulsar/pom.xml

@@ -36,7 +36,7 @@ under the License.
 	<packaging>jar</packaging>
 
 	<properties>
-		<pulsar.version>2.10.1</pulsar.version>
+		<pulsar.version>2.10.2</pulsar.version>
 
 		<!-- Test Libraries -->
 		<protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchema.java

@@ -155,6 +155,9 @@ private void writeObject(ObjectOutputStream oos) throws IOException {
             oos.writeUTF(entry.getKey());
             oos.writeUTF(entry.getValue());
         }
+
+        // Timestamp
+        oos.writeLong(schemaInfo.getTimestamp());
     }
 
     private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IOException {
@@ -177,7 +180,16 @@ private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IO
             properties.put(ois.readUTF(), ois.readUTF());
         }
 
-        this.schemaInfo = new SchemaInfoImpl(name, schemaBytes, type, properties);
+        // Timestamp
+        long timestamp = ois.readLong();
+
+        this.schemaInfo = SchemaInfoImpl.builder()
+                .name(name)
+                .schema(schemaBytes)
+                .type(type)
+                .properties(properties)
+                .timestamp(timestamp)
+                .build();
         this.schema = createSchema(schemaInfo);
     }
 

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtils.java

@@ -181,8 +181,13 @@ public static SchemaInfo encodeClassInfo(SchemaInfo schemaInfo, Class<?> typeCla
         Map<String, String> properties = new HashMap<>(schemaInfo.getProperties());
         properties.put(CLASS_INFO_PLACEHOLDER, typeClass.getName());
 
-        return new SchemaInfoImpl(
-                schemaInfo.getName(), schemaInfo.getSchema(), schemaInfo.getType(), properties);
+        return SchemaInfoImpl.builder()
+                .name(schemaInfo.getName())
+                .schema(schemaInfo.getSchema())
+                .type(schemaInfo.getType())
+                .timestamp(schemaInfo.getTimestamp())
+                .properties(properties)
+                .build();
     }
 
     @SuppressWarnings("unchecked")