[Feature] [Activemq] Added activemq sink (#7251)

.github/workflows/labeler/label-scope-conf.yml

@@ -252,6 +252,11 @@ Milvus:
       - changed-files:
           - any-glob-to-any-file: seatunnel-connectors-v2/connector-milvus/**
           - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(milvus)/**'
+activemq:
+  - all:
+      - changed-files:
+          - any-glob-to-any-file: seatunnel-connectors-v2/connector-activemq/**
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(activemq)/**'
 Zeta Rest API:
   - changed-files:
       - any-glob-to-any-file: seatunnel-engine/**/server/rest/**

config/plugin_config

@@ -86,4 +86,5 @@ connector-rocketmq
 connector-tdengine
 connector-web3j
 connector-milvus
+connector-activemq
 --end--

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

@@ -0,0 +1,123 @@
+# Activemq
+
+> Activemq sink connector
+
+## Description
+
+Used to write data to Activemq.
+
+## Key features
+
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+
+## Options
+
+|                name                 |  type   | required | default value |
+|-------------------------------------|---------|----------|---------------|
+| host                                | string  | no       | -             |
+| port                                | int     | no       | -             |
+| virtual_host                        | string  | no       | -             |
+| username                            | string  | no       | -             |
+| password                            | string  | no       | -             |
+| queue_name                          | string  | yes      | -             |
+| uri                                 | string  | yes      | -             |
+| check_for_duplicate                 | boolean | no       | -             |
+| client_id                           | boolean | no       | -             |
+| copy_message_on_send                | boolean | no       | -             |
+| disable_timeStamps_by_default       | boolean | no       | -             |
+| use_compression                     | boolean | no       | -             |
+| always_session_async                | boolean | no       | -             |
+| dispatch_async                      | boolean | no       | -             |
+| nested_map_and_list_enabled         | boolean | no       | -             |
+| warnAboutUnstartedConnectionTimeout | boolean | no       | -             |
+| closeTimeout                        | int     | no       | -             |
+
+### host [string]
+
+the default host to use for connections
+
+### port [int]
+
+the default port to use for connections
+
+### username [string]
+
+the AMQP user name to use when connecting to the broker
+
+### password [string]
+
+the password to use when connecting to the broker
+
+### uri [string]
+
+convenience method for setting the fields in an AMQP URI: host, port, username, password and virtual host
+
+### queue_name [string]
+
+the queue to write the message to
+
+### check_for_duplicate [boolean]
+
+will check for duplucate messages
+
+### client_id [string]
+
+client id
+
+### copy_message_on_send [boolean]
+
+if true, enables new JMS Message object as part of the send method
+
+### disable_timeStamps_by_default [boolean]
+
+disables timestamp for slight performance boost
+
+### use_compression [boolean]
+
+Enables the use of compression on the message’s body.
+
+### always_session_async [boolean]
+
+When true a separate thread is used for dispatching messages for each Session in the Connection.
+
+### always_sync_send [boolean]
+
+When true a MessageProducer will always use Sync sends when sending a Message
+
+### close_timeout [boolean]
+
+Sets the timeout, in milliseconds, before a close is considered complete.
+
+### dispatch_async [boolean]
+
+Should the broker dispatch messages asynchronously to the consumer
+
+### nested_map_and_list_enabled [boolean]
+
+Controls whether Structured Message Properties and MapMessages are supported
+
+### warn_about_unstarted_connection_timeout [int]
+
+The timeout, in milliseconds, from the time of connection creation to when a warning is generated
+
+## Example
+
+simple:
+
+```hocon
+sink {
+      ActiveMQ {
+          uri="tcp://localhost:61616"
+          username = "admin"
+          password = "admin"
+          queue_name = "test1"
+      }
+}
+```
+
+## Changelog
+
+### next version
+
+- Add Activemq Source Connector
+

plugin-mapping.properties

@@ -129,3 +129,4 @@ seatunnel.source.ObsFile = connector-file-obs
 seatunnel.sink.ObsFile = connector-file-obs
 seatunnel.source.Milvus = connector-milvus
 seatunnel.sink.Milvus = connector-milvus
+seatunnel.sink.ActiveMQ = connector-activemq

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

@@ -0,0 +1,65 @@
+<?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-connectors-v2</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-activemq</artifactId>
+    <name>SeaTunnel : Connectors V2 : Activemq</name>
+
+    <properties>
+        <activemq.version>5.14.5</activemq.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-client</artifactId>
+            <version>${activemq.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>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>
+
+    </dependencies>
+</project>

seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java

@@ -0,0 +1,156 @@
+/*
+ * 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.activemq.client;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.connectors.seatunnel.activemq.exception.ActivemqConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.activemq.exception.ActivemqConnectorException;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.ALWAYS_SESSION_ASYNC;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.ALWAYS_SYNC_SEND;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.CHECK_FOR_DUPLICATE;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.CLIENT_ID;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.CLOSE_TIMEOUT;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.CONSUMER_EXPIRY_CHECK_ENABLED;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.DISPATCH_ASYNC;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.NESTED_MAP_AND_LIST_ENABLED;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.PASSWORD;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.QUEUE_NAME;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.URI;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.USERNAME;
+import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqConfig.WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT;
+
+@Slf4j
+@AllArgsConstructor
+public class ActivemqClient {
+    private final ReadonlyConfig config;
+    private final ActiveMQConnectionFactory connectionFactory;
+    private final Connection connection;
+
+    public ActivemqClient(ReadonlyConfig config) {
+        this.config = config;
+        try {
+            this.connectionFactory = getConnectionFactory();
+            log.info("connection factory created");
+            this.connection = createConnection(config);
+            log.info("connection created");
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new ActivemqConnectorException(
+                    ActivemqConnectorErrorCode.CREATE_ACTIVEMQ_CLIENT_FAILED,
+                    "Error while create AMQ client ");
+        }
+    }
+
+    public ActiveMQConnectionFactory getConnectionFactory() {
+        log.info("broker url : " + config.get(URI));
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(config.get(URI));
+
+        if (config.get(ALWAYS_SESSION_ASYNC) != null) {
+            factory.setAlwaysSessionAsync(config.get(ALWAYS_SESSION_ASYNC));
+        }
+
+        if (config.get(CLIENT_ID) != null) {
+            factory.setClientID(config.get(CLIENT_ID));
+        }
+
+        if (config.get(ALWAYS_SYNC_SEND) != null) {
+            factory.setAlwaysSyncSend(config.get(ALWAYS_SYNC_SEND));
+        }
+
+        if (config.get(CHECK_FOR_DUPLICATE) != null) {
+            factory.setCheckForDuplicates(config.get(CHECK_FOR_DUPLICATE));
+        }
+
+        if (config.get(CLOSE_TIMEOUT) != null) {
+            factory.setCloseTimeout(config.get(CLOSE_TIMEOUT));
+        }
+
+        if (config.get(CONSUMER_EXPIRY_CHECK_ENABLED) != null) {
+            factory.setConsumerExpiryCheckEnabled(config.get(CONSUMER_EXPIRY_CHECK_ENABLED));
+        }
+        if (config.get(DISPATCH_ASYNC) != null) {
+            factory.setDispatchAsync(config.get(DISPATCH_ASYNC));
+        }
+
+        if (config.get(WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT) != null) {
+            factory.setWarnAboutUnstartedConnectionTimeout(
+                    config.get(WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT));
+        }
+
+        if (config.get(NESTED_MAP_AND_LIST_ENABLED) != null) {
+            factory.setNestedMapAndListEnabled(config.get(NESTED_MAP_AND_LIST_ENABLED));
+        }
+        return factory;
+    }
+
+    public void write(byte[] msg) {
+        try {
+            this.connection.start();
+            Session session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Destination destination = session.createQueue(config.get(QUEUE_NAME));
+            MessageProducer producer = session.createProducer(destination);
+            String messageBody = new String(msg, StandardCharsets.UTF_8);
+            TextMessage objectMessage = session.createTextMessage(messageBody);
+            producer.send(objectMessage);
+
+        } catch (JMSException e) {
+            throw new ActivemqConnectorException(
+                    ActivemqConnectorErrorCode.SEND_MESSAGE_FAILED,
+                    String.format(
+                            "Cannot send AMQ message %s at %s",
+                            config.get(QUEUE_NAME), config.get(CLIENT_ID)),
+                    e);
+        }
+    }
+
+    public void close() {
+        try {
+            if (connection != null) {
+                connection.close();
+            }
+        } catch (JMSException e) {
+            throw new ActivemqConnectorException(
+                    ActivemqConnectorErrorCode.CLOSE_CONNECTION_FAILED,
+                    String.format(
+                            "Error while closing AMQ connection with  %s", config.get(QUEUE_NAME)));
+        }
+    }
+
+    private Connection createConnection(ReadonlyConfig config) throws JMSException {
+        if (config.get(USERNAME) != null && config.get(PASSWORD) != null) {
+            return connectionFactory.createConnection(config.get(USERNAME), config.get(PASSWORD));
+        }
+        return connectionFactory.createConnection();
+    }
+}