[Connector-V2] Add Email sink connector

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

@@ -0,0 +1,71 @@
+# Email
+
+## Description
+
+Send the data as a file to email.
+
+ The tested email version is 1.5.6.
+
+## Options
+
+| name                     | type    | required | default value |
+|--------------------------|---------|----------|---------------|
+| email_from_address             | string  | yes      | -             |
+| email_to_address               | string  | yes      | -             |
+| email_host               | string  | yes      | -             |
+| email_transport_protocol             | string  | yes      | -             |
+| email_smtp_auth               | string  | yes      | -             |
+| email_authorization_code               | string  | yes      | -             |
+| email_message_headline             | string  | yes      | -             |
+| email_message_content               | string  | yes      | -             |
+
+
+### email_from_address [string]
+
+Sender Email Address .
+
+### email_to_address [string]
+
+Address to receive mail.
+
+### email_host [string]
+
+SMTP server to connect to.
+
+### email_transport_protocol [string]
+
+The protocol to load the session .
+
+### email_smtp_auth [string]
+
+Whether to authenticate the customer.
+
+### email_authorization_code [string]
+
+authorization code,You can obtain the authorization code from the mailbox Settings.
+
+### email_message_headline [string]
+
+The subject line of the entire message.
+
+### email_message_content [string]
+
+The body of the entire message.
+
+
+## Example
+
+```bash
+
+ EmailSink {
+      email_from_address = "xxxxxx@qq.com"
+      email_to_address = "xxxxxx@163.com"
+      email_host="smtp.qq.com"
+      email_transport_protocol="smtp"
+      email_smtp_auth="true"
+      email_authorization_code=""
+      email_message_headline=""
+      email_message_content=""
+   }
+
+```

plugin-mapping.properties

@@ -102,6 +102,7 @@ seatunnel.sink.Clickhouse = connector-clickhouse
 seatunnel.sink.ClickhouseFile = connector-clickhouse
 seatunnel.source.Jdbc = connector-jdbc
 seatunnel.sink.Jdbc = connector-jdbc
+seatunnel.sink.Email = connector-email
 seatunnel.sink.HdfsFile = connector-file-hadoop
 seatunnel.sink.LocalFile = connector-file-local
 seatunnel.source.Pulsar = connector-pulsar

pom.xml

@@ -170,6 +170,7 @@
         <mongo-spark.version>2.2.0</mongo-spark.version>
         <spark-redis.version>2.6.0</spark-redis.version>
         <commons-lang3.version>3.4</commons-lang3.version>
+        <email.version>1.5.6</email.version>
         <commons-collections4.version>4.4</commons-collections4.version>
         <maven-assembly-plugin.version>3.3.0</maven-assembly-plugin.version>
         <spark.scope>provided</spark.scope>
@@ -263,7 +264,12 @@
                 <artifactId>lz4</artifactId>
                 <version>1.3.0</version>
             </dependency>
-
+            <!--email -->
+            <dependency>
+                <groupId>com.sun.mail</groupId>
+                <artifactId>javax.mail</artifactId>
+                <version>${email.version}</version>
+            </dependency>
             <!--flink-->
             <dependency>
                 <groupId>org.apache.flink</groupId>
@@ -1322,4 +1328,4 @@
         </plugins>
     </build>
 
-</project>
+</project>

seatunnel-connectors-v2-dist/pom.xml

@@ -101,6 +101,11 @@
             <artifactId>connector-dingtalk</artifactId>
             <version>${project.version}</version>
         </dependency>
+          <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-email</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -0,0 +1,49 @@
+<?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">
+    <parent>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>connector-email</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.sun.mail</groupId>
+            <artifactId>javax.mail</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>

seatunnel-connectors-v2/connector-email/src/main/java/org/apache/seatunnel/connectors/seatunnel/email/config/EmailSinkConfig.java

@@ -0,0 +1,71 @@
+/*
+ * 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.connectors.seatunnel.email.config;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import lombok.Data;
+import lombok.NonNull;
+
+@Data
+public class EmailSinkConfig {
+
+    private static final String EMAIL_FROM_ADDRESS = "email_from_address";
+    private static final String EMAIL_TO_ADDRESS = "email_to_address";
+    private static final String EMAIL_AUTHORIZATION_CODE = "email_authorization_code";
+    private static final String EMAIL_MESSAGE_HEADLINE = "email_message_headline";
+    private static final String EMAIL_MESSAGE_CONTENT = "email_message_content";
+    private static final String EMAIL_HOST = "email_host";
+    private static final String EMAIL_TRANSPORT_PROTOCOL = "email_transport_protocol";
+    private static final String EMAIL_SMTP_AUTH = "email_smtp_auth";
+    private String emailFromAddress;
+    private String emailToAddress;
+    private String emailAuthorizationCode;
+    private String emailMessageHeadline;
+    private String emailMessageContent;
+    private String emailHost;
+    private String emailTransportProtocol;
+    private String emailSmtpAuth;
+
+    public EmailSinkConfig(@NonNull Config pluginConfig) {
+        if (pluginConfig.hasPath(EMAIL_FROM_ADDRESS)) {
+            this.emailFromAddress = pluginConfig.getString(EMAIL_FROM_ADDRESS);
+        }
+        if (pluginConfig.hasPath(EMAIL_TO_ADDRESS)) {
+            this.emailToAddress = pluginConfig.getString(EMAIL_TO_ADDRESS);
+        }
+        if (pluginConfig.hasPath(EMAIL_AUTHORIZATION_CODE)) {
+            this.emailAuthorizationCode = pluginConfig.getString(EMAIL_AUTHORIZATION_CODE);
+        }
+        if (pluginConfig.hasPath(EMAIL_MESSAGE_HEADLINE)) {
+            this.emailMessageHeadline = pluginConfig.getString(EMAIL_MESSAGE_HEADLINE);
+        }
+        if (pluginConfig.hasPath(EMAIL_MESSAGE_CONTENT)) {
+            this.emailMessageContent = pluginConfig.getString(EMAIL_MESSAGE_CONTENT);
+        }
+        if (pluginConfig.hasPath(EMAIL_HOST)) {
+            this.emailHost = pluginConfig.getString(EMAIL_HOST);
+        }
+        if (pluginConfig.hasPath(EMAIL_TRANSPORT_PROTOCOL)) {
+            this.emailTransportProtocol = pluginConfig.getString(EMAIL_TRANSPORT_PROTOCOL);
+        }
+        if (pluginConfig.hasPath(EMAIL_SMTP_AUTH)) {
+            this.emailSmtpAuth = pluginConfig.getString(EMAIL_SMTP_AUTH);
+        }
+    }
+}

seatunnel-connectors-v2/connector-email/src/main/java/org/apache/seatunnel/connectors/seatunnel/email/sink/EmailSink.java

@@ -0,0 +1,63 @@
+/*
+ * 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.connectors.seatunnel.email.sink;
+
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkWriter;
+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.connectors.seatunnel.common.sink.AbstractSimpleSink;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(SeaTunnelSink.class)
+public class EmailSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
+
+    private Config pluginConfig;
+    private SeaTunnelRowType seaTunnelRowType;
+
+    @Override
+    public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
+        this.seaTunnelRowType = seaTunnelRowType;
+    }
+
+    @Override
+    public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
+        return this.seaTunnelRowType;
+    }
+
+    @Override
+    public AbstractSinkWriter<SeaTunnelRow, Void> createWriter(SinkWriter.Context context) {
+        return new EmailSinkWriter(seaTunnelRowType, pluginConfig);
+    }
+
+    @Override
+    public String getPluginName() {
+        return "EmailSink";
+    }
+
+    @Override
+    public void prepare(Config pluginConfig) {
+        this.pluginConfig = pluginConfig;
+    }
+
+}