feat: convert the load event into interface.

Author: syhily

docker-compose.yml

@@ -32,10 +32,3 @@ services:
       - ./docker/bootstrap.sh:/pulsar/bin/bootstrap.sh
     restart: always
     command: ["/pulsar/bin/bootstrap.sh"]
-
-  flink:
-    image: streamnative/pulsar-flink:1.16.0.0
-    container_name: Flink
-    command: /bin/bash
-    depends_on:
-      - pulsar

src/main/java/io/streamnative/flink/java/dynamic/DynamicDeserializationSchema.java

@@ -0,0 +1,25 @@
+package io.streamnative.flink.java.dynamic;
+
+import io.streamnative.flink.java.models.LoadCreatedEvent;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.Collector;
+import org.apache.pulsar.client.api.Message;
+
+/**
+ * Query the schema by schema version.
+ */
+public class DynamicDeserializationSchema implements PulsarDeserializationSchema<LoadCreatedEvent> {
+    private static final long serialVersionUID = 3320218454364912622L;
+
+    @Override
+    public void deserialize(Message<byte[]> message, Collector<LoadCreatedEvent> collector) throws Exception {
+
+    }
+
+    @Override
+    public TypeInformation<LoadCreatedEvent> getProducedType() {
+        return Types.POJO(LoadCreatedEvent.class);
+    }
+}

src/main/java/io/streamnative/flink/java/generator/RandomLoadEventGenerator.java

@@ -26,7 +26,7 @@
 import net.datafaker.Faker;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 
 import java.io.Serializable;
 import java.util.List;
@@ -65,11 +65,7 @@ private LoadEvent randomLoadCreatedEvent() {
 
         return new LoadCreatedEvent()
             .setCreatedAction(faker.bigBangTheory().character())
-            .setMessages(messages)
-            .setUuid(UUID.randomUUID().toString())
-            .setId(faker.number().positive())
-            .setName(faker.name().name())
-            .setContent(faker.commerce().productName());
+            .setMessages(messages);
     }
 
     private LoadEvent randomLoadDeletedEvent() {
@@ -79,21 +75,13 @@ private LoadEvent randomLoadDeletedEvent() {
 
         return new LoadDeletedEvent()
             .setDeletedIds(ids)
-            .setErrorMsg(faker.appliance().brand())
-            .setUuid(UUID.randomUUID().toString())
-            .setId(faker.number().positive())
-            .setName(faker.name().name())
-            .setContent(faker.commerce().productName());
+            .setErrorMsg(faker.appliance().brand());
     }
 
     private LoadEvent randomLoadUpdatedEvent() {
         return new LoadUpdatedEvent()
             .setNewAction(faker.pokemon().name())
-            .setErrorMsg(faker.appliance().brand())
-            .setUuid(UUID.randomUUID().toString())
-            .setId(faker.number().positive())
-            .setName(faker.name().name())
-            .setContent(faker.commerce().productName());
+            .setErrorMsg(faker.appliance().brand());
     }
 
     @Override
@@ -103,6 +91,7 @@ public void open(RuntimeContext runtimeContext) {
 
     @Override
     public TypeInformation<LoadEvent> getType() {
-        return Types.POJO(LoadEvent.class);
+        // This is only used for demo, you can't pass an event with such type information in production environments.
+        return TypeExtractor.getForClass(LoadEvent.class);
     }
 }

src/main/java/io/streamnative/flink/java/models/LoadCreatedEvent.java

@@ -19,7 +19,6 @@
 package io.streamnative.flink.java.models;
 
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 import lombok.experimental.Accessors;
 
 import java.util.List;
@@ -29,8 +28,7 @@
  */
 @Data
 @Accessors(chain = true)
-@EqualsAndHashCode(callSuper = true)
-public class LoadCreatedEvent extends LoadEvent {
+public class LoadCreatedEvent implements LoadEvent {
     private static final long serialVersionUID = 8378154813741037238L;
 
     private String createdAction;

src/main/java/io/streamnative/flink/java/models/LoadDeletedEvent.java

@@ -19,7 +19,6 @@
 package io.streamnative.flink.java.models;
 
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 import lombok.experimental.Accessors;
 
 import java.util.List;
@@ -29,8 +28,7 @@
  */
 @Data
 @Accessors(chain = true)
-@EqualsAndHashCode(callSuper = true)
-public class LoadDeletedEvent extends LoadEvent {
+public class LoadDeletedEvent implements LoadEvent {
     private static final long serialVersionUID = -3637355499307958595L;
 
     private List<String> deletedIds;

src/main/java/io/streamnative/flink/java/models/LoadEvent.java

@@ -18,8 +18,6 @@
 
 package io.streamnative.flink.java.models;
 
-import lombok.Data;
-import lombok.experimental.Accessors;
 import org.apache.pulsar.shade.com.fasterxml.jackson.annotation.JsonSubTypes;
 import org.apache.pulsar.shade.com.fasterxml.jackson.annotation.JsonTypeInfo;
 
@@ -29,8 +27,6 @@
  * Common class for events. We should add the Jackson annotation here.
  * You should use the Pulsar-shaded Jackson annotations.
  */
-@Data
-@Accessors(chain = true)
 @JsonTypeInfo(
     use = JsonTypeInfo.Id.CLASS,
     include = JsonTypeInfo.As.PROPERTY,
@@ -41,14 +37,6 @@
     @JsonSubTypes.Type(value = LoadDeletedEvent.class, name = "LoadDeletedEvent"),
     @JsonSubTypes.Type(value = LoadUpdatedEvent.class, name = "LoadUpdatedEvent"),
 })
-public class LoadEvent implements Serializable {
-    private static final long serialVersionUID = 2895533085240328403L;
-
-    private String uuid;
-
-    private Integer id;
-
-    private String name;
-
-    private String content;
+public interface LoadEvent extends Serializable {
+    // This is an empty interface.
 }

src/main/java/io/streamnative/flink/java/models/LoadUpdatedEvent.java

@@ -19,16 +19,14 @@
 package io.streamnative.flink.java.models;
 
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 import lombok.experimental.Accessors;
 
 /**
  * The event for deleting elements.
  */
 @Data
 @Accessors(chain = true)
-@EqualsAndHashCode(callSuper = true)
-public class LoadUpdatedEvent extends LoadEvent {
+public class LoadUpdatedEvent implements LoadEvent {
     private static final long serialVersionUID = -6271238097681840919L;
 
     private String newAction;

src/main/java/io/streamnative/flink/java/polymorphic/LoadEventSeDe.java

@@ -0,0 +1,33 @@
+package io.streamnative.flink.java.polymorphic;
+
+import io.streamnative.flink.java.generator.RandomLoadEventGenerator;
+import io.streamnative.flink.java.models.LoadEvent;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.shade.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.pulsar.shade.com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * This is an example on how to serialize/deserializer the load event interface with jackson.
+ */
+public class LoadEventSeDe {
+
+    private static final ObjectMapper mapper = ObjectMapperFactory.create();
+
+    public static void main(String[] args) throws JsonProcessingException {
+        RandomLoadEventGenerator generator = new RandomLoadEventGenerator();
+        generator.open(null);
+
+        while (true) {
+            LoadEvent event = generator.generate();
+            System.out.println("The event type is: " + event.getClass().getName());
+
+            // Serialize it into JSON.
+            String json = mapper.writeValueAsString(event);
+            System.out.println("Serialize into JSON: "+ json);
+
+            // Deserializer the json into interface.
+            LoadEvent value = mapper.readValue(json, LoadEvent.class);
+            System.out.println("Deserailize the json, class type: " + value.getClass().getName() + ", content: " + value);
+        }
+    }
+}