[Improve][Connector-V2] Pulsar support user-defined schema (apache#2436)

* [Improve][Connector-V2] Pulsar support user-defined schema * import the constant
MRYOG · Sep 16, 2022 · 632901a · 632901a
1 parent 95a492d
commit 632901a
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 1 deletion.
diff --git a/seatunnel-connectors-v2/connector-pulsar/pom.xml b/seatunnel-connectors-v2/connector-pulsar/pom.xml
@@ -39,6 +39,19 @@
             <artifactId>seatunnel-api</artifactId>
             <version>${project.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-json</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <!-- Pulsar testing environment -->
 
         <dependency>

diff --git a/...r/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java b/...r/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
@@ -46,9 +46,11 @@
 import org.apache.seatunnel.api.source.SourceReader;
 import org.apache.seatunnel.api.source.SourceSplitEnumerator;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.config.CheckConfigUtil;
 import org.apache.seatunnel.common.config.CheckResult;
 import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.connectors.seatunnel.common.schema.SeatunnelSchema;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarAdminConfig;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarClientConfig;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConsumerConfig;
@@ -64,6 +66,7 @@
 import org.apache.seatunnel.connectors.seatunnel.pulsar.source.enumerator.discoverer.TopicPatternDiscoverer;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.source.reader.PulsarSourceReader;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.seatunnel.format.json.JsonDeserializationSchema;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
@@ -222,7 +225,9 @@ private void setPartitionDiscoverer(Config config) {
 
     private void setDeserialization(Config config) {
         String format = config.getString("format");
-        // TODO: json format
+        // TODO: format SPI
+        SeaTunnelRowType rowType = SeatunnelSchema.buildWithConfig(config.getConfig(SeatunnelSchema.SCHEMA)).getSeaTunnelRowType();
+        deserialization = (DeserializationSchema<T>) new JsonDeserializationSchema(false, false, rowType);
     }
 
     @Override