[Improve][Connector][PulsarSource]Improve pulsar deserialization

docs/en/connector-v2/source/pulsar.md

@@ -36,6 +36,8 @@ Source connector for Apache Pulsar.
 | cursor.stop.mode         | Enum    | No       | NEVER         |
 | cursor.stop.timestamp    | Long    | No       | -             |
 | schema                   | config  | No       | -             |
+| format                   | String  | no       | json          |
+| field_delimiter          | String  | no       | ,             |
 | common-options           |         | no       | -             |
 
 ### topic [String]
@@ -126,6 +128,15 @@ Stop from the specified epoch timestamp (in milliseconds).
 
 ### schema [Config]
 
+## format
+
+Data format. The default format is json. Optional text format. The default field separator is ", ".
+If you customize the delimiter, add the "field_delimiter" option.
+
+## field_delimiter
+
+Customize the field delimiter for data format.
+
 #### fields [Config]
 
 the schema fields of upstream data.
@@ -154,3 +165,7 @@ source {
 
 - Add Pulsar Source Connector
 
+### Next Version
+
+- Improve pulsar deserialization([3990](https://github.com/apache/incubator-seatunnel/pull/3990))
+

seatunnel-connectors-v2/connector-pulsar/pom.xml

@@ -41,6 +41,12 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-text</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
             <artifactId>connector-common</artifactId>

seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/SourceProperties.java

@@ -26,6 +26,13 @@ public class SourceProperties {
     private static final Integer DEFAULT_POLL_TIMEOUT = 100;
     private static final Long DEFAULT_POLL_INTERVAL = 50L;
     private static final Integer DEFAULT_POLL_BATCH_SIZE = 500;
+    /** The default data format is JSON */
+    public static final String DEFAULT_FORMAT = "json";
+
+    public static final String TEXT_FORMAT = "text";
+
+    /** The default field delimiter is “,” */
+    public static final String DEFAULT_FIELD_DELIMITER = ",";
 
     // --------------------------------------------------------------------------------------------
     // The configuration for ClientConfigurationData part.
@@ -167,8 +174,20 @@ public class SourceProperties {
                     .longType()
                     .noDefaultValue()
                     .withDescription("Stop from the specified epoch timestamp (in milliseconds)");
+    public static final Option<String> FORMAT =
+            Options.key("format")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Data format. The default format is json. Optional text format. The default field separator is \", \". "
+                                    + "If you customize the delimiter, add the \"field_delimiter\" option.");
+
+    public static final Option<String> FIELD_DELIMITER =
+            Options.key("field_delimiter")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Customize the field delimiter for data format.");
 
-    /** Startup mode for the pulsar consumer, see {@link #CURSOR_STARTUP_MODE}. */
     public enum StartMode {
         /** Start from the earliest cursor possible. */
         EARLIEST,

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/pom.xml

@@ -0,0 +1,103 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connector-v2-e2e</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-pulsar-e2e</artifactId>
+
+    <properties>
+        <pulsar.version>2.8.0</pulsar.version>
+        <testcontainers.version>1.17.6</testcontainers.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-e2e-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-pulsar</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-json</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-text</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client-all</artifactId>
+            <version>${pulsar.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.pulsar</groupId>
+                    <artifactId>pulsar-package-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-console</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-assert</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>3.23.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>pulsar</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>jul-to-slf4j</artifactId>
+            <version>${slf4j.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarIT.java

@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.connector.pulsar;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.format.text.TextSerializationSchema;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.UUID;
+
+import static java.time.temporal.ChronoUnit.SECONDS;
+
+/** Start / stop a Pulsar cluster. */
+@Slf4j
+public class PulsarIT extends TestSuiteBase implements TestResource {
+
+    private PulsarContainer pulsarService;
+
+    private String serviceUrl;
+
+    private String adminUrl;
+
+    private String zkUrl;
+
+    private ClientConfigurationData clientConfigurationData = new ClientConfigurationData();
+
+    private ConsumerConfigurationData<byte[]> consumerConfigurationData =
+            new ConsumerConfigurationData<>();
+
+    private PulsarAdmin pulsarAdmin;
+
+    private PulsarClient pulsarClient;
+
+    private static final String TOPIC = "persistent://public/default/test";
+
+    @BeforeAll
+    @Override
+    public void startUp() throws Exception {
+
+        log.info("Starting PulsarService ");
+        pulsarService = new PulsarContainer();
+        pulsarService.addExposedPort(2181);
+        pulsarService.waitingFor(
+                new HttpWaitStrategy()
+                        .forPort(PulsarContainer.BROKER_HTTP_PORT)
+                        .forStatusCode(200)
+                        .forPath("/admin/v2/namespaces/public/default")
+                        .withStartupTimeout(Duration.of(400, SECONDS)));
+        pulsarService.start();
+        pulsarService.followOutput(new Slf4jLogConsumer(log));
+        serviceUrl = pulsarService.getPulsarBrokerUrl();
+        adminUrl = pulsarService.getHttpServiceUrl();
+        zkUrl = "localhost:" + pulsarService.getMappedPort(2181);
+        clientConfigurationData.setServiceUrl(serviceUrl);
+        consumerConfigurationData.setSubscriptionMode(SubscriptionMode.NonDurable);
+        consumerConfigurationData.setSubscriptionType(SubscriptionType.Exclusive);
+        consumerConfigurationData.setSubscriptionName("flink-" + UUID.randomUUID());
+        log.info("serviceUrl:{}", serviceUrl);
+        log.info("adminUrl:{}", adminUrl);
+        log.info("zkUrl:{}", zkUrl);
+
+        ClientBuilder builder = PulsarClient.builder();
+        builder.serviceUrl(serviceUrl);
+        pulsarClient = builder.build();
+
+        TextSerializationSchema serializer =
+                TextSerializationSchema.builder()
+                        .seaTunnelRowType(seaTunnelRowType)
+                        .delimiter(",")
+                        .build();
+        generateTestData(row -> new String(serializer.serialize(row)), 0, 2);
+
+        log.info("Successfully started PulsarService");
+    }
+
+    @AfterAll
+    @Override
+    public void tearDown() throws Exception {
+        log.info("-------------------------------------------------------------------------");
+        log.info("    Shut down PulsarService ");
+        log.info("-------------------------------------------------------------------------");
+
+        if (pulsarService != null) {
+            pulsarService.stop();
+        }
+        if (pulsarAdmin != null) {
+            pulsarAdmin.close();
+        }
+
+        log.info("-------------------------------------------------------------------------");
+        log.info("    PulsarService finished");
+        log.info("-------------------------------------------------------------------------");
+    }
+
+    @TestTemplate
+    public void testSourcePulsarTextToConsole(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult =
+                container.executeJob("/pulsarsource_text_to_console.conf");
+        log.info("execResult.getExitCode:{}", execResult.getExitCode());
+        log.info("execResult.getStdout:{}", execResult.getStdout());
+        log.info("execResult.getStderr:{}", execResult.getStderr());
+    }
+
+    private void generateTestData(ProducerRecordConverter converter, int start, int end)
+            throws PulsarClientException {
+        try (Producer<String> producer =
+                pulsarClient.newProducer(Schema.STRING).topic(TOPIC).create(); ) {
+            for (int i = start; i < end; i++) {
+                SeaTunnelRow row =
+                        new SeaTunnelRow(new Object[] {Long.valueOf(i), "pulsarsource-test" + i});
+                String producerRecord = converter.convert(row);
+                producer.send(producerRecord);
+            }
+        }
+    }
+
+    interface ProducerRecordConverter {
+        String convert(SeaTunnelRow row);
+    }
+
+    @SuppressWarnings("checkstyle:InnerTypeLast")
+    SeaTunnelRowType seaTunnelRowType =
+            new SeaTunnelRowType(
+                    new String[] {"id", "c_string"},
+                    new SeaTunnelDataType[] {BasicType.LONG_TYPE, BasicType.STRING_TYPE});
+}