[Feature][Connector-V2] Add sentry sink connector #2244

docs/en/connector-v2/sink/Sentry.md

@@ -0,0 +1,53 @@
+# Sentry
+
+## Description
+
+Write message to Sentry.
+
+## Options
+
+| name                       | type    | required | default value |
+|----------------------------|---------|----------| ------------- |
+| dsn                        | string  | yes      | -             |
+| env                        | string  | no       | -             |
+| release                    | string  | no       | -             |
+| cacheDirPath               | string  | no       | -             |
+| enableExternalConfiguration | boolean | no       | -             |
+| maxCacheItems              | number  | no       | -             |
+| flushTimeoutMills          | number  | no       | -             |
+| maxQueueSize               | number  | no       | -             |
+### dsn [string]
+
+The DSN tells the SDK where to send the events to.
+
+### env [string]
+specify the environment
+
+### release [string]
+specify the release
+
+### cacheDirPath [string]
+the cache dir path for caching offline events
+
+### enableExternalConfiguration [boolean]
+if loading properties from external sources is enabled.
+
+### maxCacheItems [number]
+The max cache items for capping the number of events Default is 30
+
+### flushTimeoutMillis [number]
+Controls how many seconds to wait before flushing down. Sentry SDKs cache events from a background queue and this queue is given a certain amount to drain pending events Default is 15000 = 15s
+
+### maxQueueSize [number]
+Max queue size before flushing events/envelopes to the disk
+
+## Example
+```
+  Sentry {
+    dsn = "https://xxx@sentry.xxx.com:9999/6"
+    enableExternalConfiguration = true
+    maxCacheItems = 1000
+    env = prod
+  }
+
+```

plugin-mapping.properties

@@ -122,4 +122,4 @@ seatunnel.sink.Neo4j = connector-neo4j
 seatunnel.sink.FtpFile = connector-file-ftp
 seatunnel.sink.Socket = connector-socket
 seatunnel.sink.DataHub = connector-datahub
-
+seatunnel.sink.Sentry = connector-sentry

seatunnel-connectors-v2-dist/pom.xml

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

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

@@ -0,0 +1,58 @@
+<?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-sentry</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+<!--        <dependency>-->
+<!--            <groupId>org.apache.commons</groupId>-->
+<!--            <artifactId>commons-lang3</artifactId>-->
+<!--        </dependency>-->
+
+        <dependency>
+            <groupId>io.sentry</groupId>
+            <artifactId>sentry-logback</artifactId>
+            <version>5.0.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>2.1.3-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>

seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/sink/SentryConfig.java

@@ -0,0 +1,32 @@
+/*
+ * 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.sentry.sink;
+
+public class SentryConfig {
+
+    public static final String SENTRY = "sentry";
+    public static final String DSN = "dsn";
+    public static final String ENV = "env";
+    public static final String RELEASE = "release";
+    public static final String CACHE_DIRPATH = "cacheDirPath";
+    public static final String ENABLE_EXTERNAL_CONFIGURATION = "enableExternalConfiguration";
+    public static final String MAX_CACHEITEMS = "maxCacheItems";
+    public static final String FLUSH_TIMEOUTMILLIS = "flushTimeoutMillis";
+    public static final String MAX_QUEUESIZE = "maxQueueSize";
+
+}

seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/sink/SentrySink.java

@@ -0,0 +1,73 @@
+/*
+ * 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.sentry.sink;
+
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkWriter.Context;
+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.common.constants.PluginType;
+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;
+
+import java.io.IOException;
+
+/**
+ * @description: SentrySink class
+ */
+@AutoService(SeaTunnelSink.class)
+public class SentrySink extends AbstractSimpleSink<SeaTunnelRow, SentrySinkState> {
+
+    private SeaTunnelRowType seaTunnelRowType;
+    private Config pluginConfig;
+    @Override
+    public String getPluginName() {
+        return SentryConfig.SENTRY;
+    }
+
+    @Override
+    public void prepare(Config pluginConfig) throws PrepareFailException {
+        if (!pluginConfig.hasPath(SentryConfig.DSN)) {
+            throw new PrepareFailException(getPluginName(), PluginType.SINK,
+                    String.format("Config must include column : %s", SentryConfig.DSN));
+        }
+
+        this.pluginConfig = pluginConfig;
+    }
+
+    @Override
+    public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
+        this.seaTunnelRowType = seaTunnelRowType;
+    }
+
+    @Override
+    public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
+        return this.seaTunnelRowType;
+    }
+
+    @Override
+    public AbstractSinkWriter<SeaTunnelRow, SentrySinkState> createWriter(Context context) throws IOException {
+        return new SentrySinkWriter(seaTunnelRowType, context, pluginConfig);
+    }
+}

seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/sink/SentrySinkState.java

@@ -0,0 +1,23 @@
+/*
+ * 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.sentry.sink;
+
+import java.io.Serializable;
+
+public class SentrySinkState implements Serializable {
+}

seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/sink/SentrySinkWriter.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.connectors.seatunnel.sentry.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import io.sentry.Sentry;
+import io.sentry.SentryOptions;
+
+import java.io.IOException;
+
+/**
+ * @description: SentrySinkWriter class
+ */
+
+public class SentrySinkWriter extends AbstractSinkWriter<SeaTunnelRow, SentrySinkState> {
+    private SeaTunnelRowType seaTunnelRowType;
+    public SentrySinkWriter(SeaTunnelRowType seaTunnelRowType,
+                            SinkWriter.Context context,
+                            Config pluginConfig) {
+        SentryOptions options = new SentryOptions();
+        options.setDsn(pluginConfig.getString(SentryConfig.DSN));
+        if (pluginConfig.hasPath(SentryConfig.ENV)){
+            options.setEnvironment(pluginConfig.getString(SentryConfig.ENV));
+        }
+        if (pluginConfig.hasPath(SentryConfig.RELEASE)){
+            options.setRelease(pluginConfig.getString(SentryConfig.RELEASE));
+        }
+        if (pluginConfig.hasPath(SentryConfig.CACHE_DIRPATH)){
+            options.setCacheDirPath(pluginConfig.getString(SentryConfig.CACHE_DIRPATH));
+        }
+        if (pluginConfig.hasPath(SentryConfig.MAX_CACHEITEMS)){
+            options.setMaxCacheItems(pluginConfig.getInt(SentryConfig.MAX_CACHEITEMS));
+        }
+        if (pluginConfig.hasPath(SentryConfig.MAX_QUEUESIZE)){
+            options.setMaxQueueSize(pluginConfig.getInt(SentryConfig.MAX_QUEUESIZE));
+        }
+        if (pluginConfig.hasPath(SentryConfig.FLUSH_TIMEOUTMILLIS)){
+            options.setFlushTimeoutMillis(pluginConfig.getLong(SentryConfig.FLUSH_TIMEOUTMILLIS));
+        }
+        if (pluginConfig.hasPath(SentryConfig.ENABLE_EXTERNAL_CONFIGURATION)){
+            options.setEnableExternalConfiguration(pluginConfig.getBoolean(SentryConfig.ENABLE_EXTERNAL_CONFIGURATION));
+        }
+        Sentry.init(options);
+    }
+
+    @Override
+    public void write(SeaTunnelRow element) throws IOException {
+        Sentry.captureMessage(element.toString());
+    }
+
+    @Override
+    public void close() throws IOException {
+        Sentry.close();
+    }
+
+}

seatunnel-connectors-v2/pom.xml

@@ -51,6 +51,7 @@
         <module>connector-iotdb</module>
         <module>connector-neo4j</module>
         <module>connector-datahub</module>
+        <module>connector-sentry</module>
     </modules>
 
     <dependencies>

seatunnel-examples/seatunnel-flink-connector-v2-example/pom.xml

@@ -82,7 +82,11 @@
             <version>${project.version}</version>
         </dependency>
         <!--   seatunnel connectors   -->
-
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-sentry</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <!--flink-->
         <dependency>
             <groupId>org.apache.flink</groupId>