add datasource-pulsar

seatunnel-datasource/seatunnel-datasource-client/src/main/java/org/apache/seatunnel/datasource/classloader/DatasourceLoadConfig.java

@@ -75,6 +75,8 @@ public class DatasourceLoadConfig {
                 "org.apache.seatunnel.datasource.plugin.tidb.jdbc.TidbJdbcDataSourceFactory");
         classLoaderFactoryName.put(
                 "KAFKA", "org.apache.seatunnel.datasource.plugin.kafka.KafkaDataSourceFactory");
+        classLoaderFactoryName.put(
+                "PULSAR", "org.apache.seatunnel.datasource.plugin.pulsar.PulsarDataSourceFactory");
         classLoaderFactoryName.put(
                 "MYSQL-CDC",
                 "org.apache.seatunnel.datasource.plugin.cdc.mysql.MysqlCDCDataSourceFactory");
@@ -108,6 +110,7 @@ public class DatasourceLoadConfig {
         classLoaderJarName.put("S3", "datasource-s3-");
         classLoaderJarName.put("HIVE", "datasource-hive-");
         classLoaderJarName.put("KAFKA", "datasource-kafka-");
+        classLoaderJarName.put("PULSAR", "datasource-pulsar-");
         classLoaderJarName.put("STARROCKS", "datasource-starrocks-");
         classLoaderJarName.put("S3-REDSHIFT", "datasource-s3redshift-");
         classLoaderJarName.put("JDBC-STARROCKS", "datasource-jdbc-starrocks-");
@@ -124,6 +127,7 @@ public class DatasourceLoadConfig {
                     "JDBC-SQLServer",
                     "JDBC-TiDB",
                     "Kafka",
+                    "Pulsar",
                     "MySQL-CDC",
                     "S3",
                     "SqlServer-CDC",

seatunnel-datasource/seatunnel-datasource-plugins/datasource-pulsar/pom.xml

@@ -0,0 +1,85 @@
+<?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-datasource-plugins</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>datasource-pulsar</artifactId>
+
+    <properties>
+        <pulsar.version>2.11.0</pulsar.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>datasource-plugins-api</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.commons</groupId>
+                    <artifactId>common-lang3</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/com.google.auto.service/auto-service -->
+        <dependency>
+            <groupId>com.google.auto.service</groupId>
+            <artifactId>auto-service</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client-all</artifactId>
+            <version>${pulsar.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <configuration>
+                    <skip>${e2e.dependency.skip}</skip>
+                    <appendOutput>true</appendOutput>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <scm>
+        <tag>1.0.0</tag>
+    </scm>
+</project>

seatunnel-datasource/seatunnel-datasource-plugins/datasource-pulsar/src/main/java/org/apache/seatunnel/datasource/plugin/pulsar/PulsarDataSourceChannel.java

@@ -0,0 +1,125 @@
+/*
+ * 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.datasource.plugin.pulsar;
+
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.datasource.plugin.api.DataSourceChannel;
+import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
+import org.apache.seatunnel.datasource.plugin.api.model.TableField;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PulsarDataSourceChannel implements DataSourceChannel {
+
+    private static final String TENANT = "public";
+    private static final String DATABASE = "public/default";
+    private static final int DEFAULT_TIMEOUT_OPTIONS =
+            PulsarAdminImpl.DEFAULT_CONNECT_TIMEOUT_SECONDS;
+
+    @Override
+    public OptionRule getDataSourceOptions(@NonNull String pluginName) {
+        return PulsarOptionRule.optionRule();
+    }
+
+    @Override
+    public OptionRule getDatasourceMetadataFieldsByDataSourceName(@NonNull String pluginName) {
+        return PulsarOptionRule.metadataRule();
+    }
+
+    @Override
+    public List<String> getTables(
+            @NonNull String pluginName,
+            Map<String, String> requestParams,
+            String database,
+            Map<String, String> option) {
+        checkArgument(StringUtils.equalsIgnoreCase(database, DATABASE), "database must be default");
+        try (PulsarAdmin pulsarAdmin = createPulsarAdmin(requestParams)) {
+            return pulsarAdmin.topics().getList(database);
+        } catch (Exception ex) {
+            throw new DataSourcePluginException(
+                    "check Pulsar connectivity failed, " + ex.getMessage(), ex);
+        }
+    }
+
+    @Override
+    public List<String> getDatabases(
+            @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
+        try (PulsarAdmin pulsarAdmin = createPulsarAdmin(requestParams)) {
+            return pulsarAdmin.namespaces().getNamespaces(TENANT);
+        } catch (Exception ex) {
+            throw new DataSourcePluginException(
+                    "check Pulsar connectivity failed, " + ex.getMessage(), ex);
+        }
+    }
+
+    @Override
+    public boolean checkDataSourceConnectivity(
+            @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
+        try (PulsarAdmin pulsarAdmin = createPulsarAdmin(requestParams)) {
+            // just test the connection
+            return CollectionUtils.isNotEmpty(pulsarAdmin.worker().getCluster());
+        } catch (Exception ex) {
+            throw new DataSourcePluginException(
+                    "check Pulsar connectivity failed, " + ex.getMessage(), ex);
+        }
+    }
+
+    @Override
+    public List<TableField> getTableFields(
+            @NonNull String pluginName,
+            @NonNull Map<String, String> requestParams,
+            @NonNull String database,
+            @NonNull String table) {
+        checkArgument(StringUtils.equalsIgnoreCase(database, DATABASE), "database must be default");
+        return Collections.emptyList();
+    }
+
+    @Override
+    public Map<String, List<TableField>> getTableFields(
+            @NonNull String pluginName,
+            @NonNull Map<String, String> requestParams,
+            @NonNull String database,
+            @NonNull List<String> tables) {
+        checkArgument(StringUtils.equalsIgnoreCase(database, DATABASE), "database must be default");
+        return Collections.emptyMap();
+    }
+
+    private PulsarAdmin createPulsarAdmin(Map<String, String> requestParams)
+            throws PulsarClientException {
+        return PulsarAdmin.builder()
+                .serviceHttpUrl(requestParams.get("serviceHttpUrl"))
+                .authentication(
+                        requestParams.get("authPluginClassName"),
+                        requestParams.get("authParamsString"))
+                .build();
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-pulsar/src/main/java/org/apache/seatunnel/datasource/plugin/pulsar/PulsarDataSourceFactory.java

@@ -0,0 +1,58 @@
+/*
+ * 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.datasource.plugin.pulsar;
+
+import org.apache.seatunnel.datasource.plugin.api.DataSourceChannel;
+import org.apache.seatunnel.datasource.plugin.api.DataSourceFactory;
+import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginInfo;
+import org.apache.seatunnel.datasource.plugin.api.DatasourcePluginTypeEnum;
+
+import com.google.auto.service.AutoService;
+import com.google.common.collect.Sets;
+
+import java.util.Set;
+
+@AutoService(DataSourceFactory.class)
+public class PulsarDataSourceFactory implements DataSourceFactory {
+
+    public static final String PULSAR_PLUGIN_NAME = "Pulsar";
+    public static final String PULSAR_PLUGIN_ICON = "pulsar";
+    public static final String PULSAR_PLUGIN_VERSION = "1.0.0";
+
+    @Override
+    public String factoryIdentifier() {
+        return PULSAR_PLUGIN_NAME;
+    }
+
+    @Override
+    public Set<DataSourcePluginInfo> supportedDataSources() {
+        return Sets.newHashSet(
+                DataSourcePluginInfo.builder()
+                        .name(PULSAR_PLUGIN_NAME)
+                        .icon(PULSAR_PLUGIN_ICON)
+                        .version(PULSAR_PLUGIN_VERSION)
+                        .supportVirtualTables(true)
+                        .type(DatasourcePluginTypeEnum.NO_STRUCTURED.getCode())
+                        .build());
+    }
+
+    @Override
+    public DataSourceChannel createChannel() {
+        return new PulsarDataSourceChannel();
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-pulsar/src/main/java/org/apache/seatunnel/datasource/plugin/pulsar/PulsarOptionRule.java

@@ -0,0 +1,77 @@
+/*
+ * 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.datasource.plugin.pulsar;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+
+import java.util.Map;
+
+public class PulsarOptionRule {
+
+    public static final Option<String> SERVICE_URL =
+            Options.key("serviceUrl")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Pulsar cluster HTTP URL to connect to a broker.");
+    public static final Option<String> TOPIC =
+            Options.key("topicName")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Pulsar topic name.");
+
+    public static final Option<Boolean> PATTERN =
+            Options.key("topicsPattern")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Topic pattern.");
+
+    public static final Option<Map<String, String>> PULSAR_CONFIG =
+            Options.key("pulsar.config")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "{\n"
+                                    + "numListenerThreads=2\n"
+                                    + "webSocketServiceEnabled=true\n"
+                                    + "operationTimeoutMs=30000\n"
+                                    + "proxyServiceUrl=null\n"
+                                    + "readTimeoutMs=60000\n"
+                                    + "useTcpNoDelay=true\n"
+                                    + "useTls=false\n"
+                                    + "concurrentLookupRequest=5000\n"
+                                    + "connectionMaxIdleSeconds=180\n"
+                                    + "connectionTimeoutMs=10000\n"
+                                    + "connectionsPerBroker=1\n"
+                                    + "description=1\n"
+                                    + "enableTransaction=false\n"
+                                    + "initialBackoffIntervalNanos=100000000\n"
+                                    + "sendTimeoutMs=30000\n"
+                                    + "batchingEnabled=true\n"
+                                    + "batchingMaxMessages=1000\n"
+                                    + "}");
+
+    public static OptionRule optionRule() {
+        return OptionRule.builder().required(SERVICE_URL).optional(PULSAR_CONFIG).build();
+    }
+
+    public static OptionRule metadataRule() {
+        return OptionRule.builder().required(TOPIC).optional(PATTERN).build();
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/pom.xml

@@ -41,6 +41,7 @@
         <module>datasource-jdbc-sqlserver</module>
         <module>datasource-jdbc-starrocks</module>
         <module>datasource-mysql-cdc</module>
+        <module>datasource-pulsar</module>
         <module>datasource-s3</module>
         <module>datasource-sqlserver-cdc</module>
         <module>datasource-jdbc-tidb</module>

seatunnel-server/seatunnel-app/src/main/bin/download_datasource.sh

@@ -39,6 +39,7 @@ datasource_list=(
   "datasource-jdbc-starrocks"
   "datasource-jdbc-tidb"
   "datasource-kafka"
+  "datasource-pulsar"
   "datasource-mysql-cdc"
   "datasource-s3"
   "datasource-sqlserver-cdc"