This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12698-372d3cd7b4d7715eb6fc811c1b43e23cb6309c4d in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 83603e19b7b09d6114c6f6c9e89bfdafe2796183 Author: chovygo <[email protected]> AuthorDate: Sun Oct 11 13:02:02 2026 +0000 [Improve] [Connector-V2] Improve ActiveMQ sink: fix resource leak and add config options (#12698) Co-authored-by: zhaowei <[email protected]> --- docs/en/connectors/sink/Activemq.md | 11 + docs/zh/connectors/sink/Activemq.md | 19 +- .../seatunnel/activemq/client/ActivemqClient.java | 87 +++- .../activemq/config/ActivemqSinkOptions.java | 77 ++++ .../activemq/sink/ActivemqSinkFactory.java | 20 +- .../seatunnel/activemq/ActivemqFactoryTest.java | 9 + .../activemq/client/ActivemqClientTest.java | 510 +++++++++++++++++++++ 7 files changed, 720 insertions(+), 13 deletions(-) diff --git a/docs/en/connectors/sink/Activemq.md b/docs/en/connectors/sink/Activemq.md index c3aa17156c..834b6063e0 100644 --- a/docs/en/connectors/sink/Activemq.md +++ b/docs/en/connectors/sink/Activemq.md @@ -38,6 +38,15 @@ For consuming queues, see the separate [ActiveMQ source](../source/Activemq.md). | nested_map_and_list_enabled | boolean | no | - | Whether structured message properties and `MapMessage` entries can contain nested `Map` and `List` objects. | | warn_about_unstarted_connection_timeout | int | no | - | Timeout in milliseconds before ActiveMQ warns that a connection was not started correctly. Set a value less than `0` to disable the warning in the ActiveMQ client. | | consumer_expiry_check_enabled | boolean | no | - | Whether the ActiveMQ client checks message expiration in each `MessageConsumer` before dispatching messages. | +| max_thread_pool_size | int | no | 1000 | Maximum thread pool size for the ActiveMQ connection factory's internal thread pool used for dispatching messages. Defaults to `1000`. | +| send_timeout | int | no | 0 | Send timeout in milliseconds for the ActiveMQ connection factory. A value of `0` means no timeout. Defaults to `0`. | +| use_compression | boolean | no | false | Whether message bodies are compressed before being sent to the broker. Defaults to `false`. | +| connect_response_timeout | int | no | 0 | Timeout in milliseconds for waiting for the broker's response during connection establishment. A value of `0` means no timeout. Defaults to `0`. | +| delivery_mode | int | no | 2 | JMS delivery mode: `1` for NON_PERSISTENT or `2` for PERSISTENT. Defaults to `2` (PERSISTENT). | +| priority | int | no | 4 | JMS message priority from 0 (lowest) to 9 (highest). Defaults to `4`. | +| producer_window_size | int | no | 0 | Flow control window size in bytes for async sends. A value of `0` means no flow control. Defaults to `0`. | +| time_to_live | long | no | 0 | Message time-to-live in milliseconds. A value of `0` means never expires. Defaults to `0`. | +| use_async_send | boolean | no | false | When true, sends return without waiting for broker confirmation. Defaults to `false`. | ## Notes @@ -46,6 +55,8 @@ For consuming queues, see the separate [ActiveMQ source](../source/Activemq.md). - The connector writes each SeaTunnel row as one JSON text message to `queue_name`. There is no separate `format` option for this sink. - Configure the broker address with `uri`. `host` and `port` are not ActiveMQ sink options. - Use any SeaTunnel source before this sink. The ActiveMQ connector only controls how the final rows are sent to the queue. +- When `use_async_send = true` or `delivery_mode = 1` (NON_PERSISTENT), the ActiveMQ client sends without waiting for broker confirmation and this sink does not register an `ExceptionListener`, so a failed send may not be surfaced to the job. Use these options only when acceptable message loss is tolerable. +- The factory-level options (`max_thread_pool_size`, `send_timeout`, `use_compression`, `connect_response_timeout`, `producer_window_size`, `use_async_send`) are only applied when explicitly set. When not set, the value from the matching `jms.*` parameter in the `uri` (if present) or the ActiveMQ client default is used. ## Example diff --git a/docs/zh/connectors/sink/Activemq.md b/docs/zh/connectors/sink/Activemq.md index 7cba3be4e3..5ce6ece4e5 100644 --- a/docs/zh/connectors/sink/Activemq.md +++ b/docs/zh/connectors/sink/Activemq.md @@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-activemq.md'; > ActiveMQ Sink 连接器 +## 支持的引擎 + +> Spark<br/> +> Flink<br/> +> SeaTunnel Zeta<br/> + ## 描述 用于把 SeaTunnel 数据写入 ActiveMQ 队列。每一行数据都会被序列化成一条 JSON 文本消息。 @@ -31,7 +37,16 @@ import ChangeLog from '../changelog/connector-activemq.md'; | dispatch_async | boolean | 否 | - | Broker 是否异步分发消息。 | | nested_map_and_list_enabled | boolean | 否 | - | 是否允许结构化消息属性和 `MapMessage` 条目中包含嵌套的 `Map`、`List` 对象。 | | warn_about_unstarted_connection_timeout | int | 否 | - | 连接没有正确启动时，ActiveMQ 客户端发出警告前等待的毫秒数。设置为小于 `0` 的值可以关闭这个警告。 | -| consumer_expiry_check_enabled | boolean | 否 | - | 是否在每个 `MessageConsumer` 分发消息前检查消息是否已经过期。 | +| consumer_expiry_check_enabled | boolean | 否 | - | 是否在每个 `MessageConsumer` 分发消息前检查消息是否已经过期。 | +| max_thread_pool_size | int | 否 | 1000 | ActiveMQ 连接工厂内部用于消息分发的线程池最大大小。默认值为 `1000`。 | +| send_timeout | int | 否 | 0 | ActiveMQ 连接工厂的发送超时时间，单位毫秒。值为 `0` 表示不超时。默认值为 `0`。 | +| use_compression | boolean | 否 | false | 是否在发送消息前对消息体进行压缩。默认值为 `false`。 | +| connect_response_timeout | int | 否 | 0 | 连接建立时等待 Broker 响应的超时时间，单位毫秒。值为 `0` 表示不超时。默认值为 `0`。 | +| delivery_mode | int | 否 | 2 | JMS 投递模式：`1` 为非持久化，`2` 为持久化。默认值为 `2`（持久化）。 | +| priority | int | 否 | 4 | JMS 消息优先级，0（最低）到 9（最高）。默认值为 `4`。 | +| producer_window_size | int | 否 | 0 | 异步发送时的流控窗口大小，单位字节。值为 `0` 表示不限流。默认值为 `0`。 | +| time_to_live | long | 否 | 0 | 消息存活时间，单位毫秒。值为 `0` 表示永不过期。默认值为 `0`。 | +| use_async_send | boolean | 否 | false | 为 true 时发送不等 Broker 确认直接返回，提升吞吐量。默认值为 `false`。 | ## 注意事项 @@ -40,6 +55,8 @@ import ChangeLog from '../changelog/connector-activemq.md'; - 连接器会把每一行 SeaTunnel 数据作为一条 JSON 文本消息写入 `queue_name`，当前没有单独的 `format` 配置。 - Broker 地址请使用 `uri` 配置，`host` 和 `port` 不是 ActiveMQ Sink 的配置项。 - 该 Sink 前面可以接任意 SeaTunnel Source。ActiveMQ 连接器只负责把最终的数据行发送到队列。 +- 当 `use_async_send = true` 或 `delivery_mode = 1`（非持久化）时，ActiveMQ 客户端发送消息后不等 Broker 确认即返回，且本连接器未注册 `ExceptionListener`，因此发送失败可能不会通知到作业。仅在可接受消息丢失的场景下使用这些选项。 +- 工厂级选项（`max_thread_pool_size`、`send_timeout`、`use_compression`、`connect_response_timeout`、`producer_window_size`、`use_async_send`）仅在显式配置时生效。未配置时，使用 `uri` 中对应的 `jms.*` 参数（如果存在）或 ActiveMQ 客户端默认值。 ## 示例 diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java index d351344446..b470a30ac2 100644 --- a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java @@ -39,13 +39,22 @@ import static org.apache.seatunnel.connectors.seatunnel.activemq.config.Activemq import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CHECK_FOR_DUPLICATE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CLIENT_ID; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CLOSE_TIMEOUT; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CONNECT_RESPONSE_TIMEOUT; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CONSUMER_EXPIRY_CHECK_ENABLED; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.DELIVERY_MODE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.DISPATCH_ASYNC; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.MAX_THREAD_POOL_SIZE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.NESTED_MAP_AND_LIST_ENABLED; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PASSWORD; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PRIORITY; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PRODUCER_WINDOW_SIZE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.QUEUE_NAME; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.SEND_TIMEOUT; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.TIME_TO_LIVE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.URI; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.USERNAME; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.USE_ASYNC_SEND; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.USE_COMPRESSION; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT; @Slf4j @@ -53,17 +62,44 @@ public class ActivemqClient { private final ReadonlyConfig config; private final ActiveMQConnectionFactory connectionFactory; private final Connection connection; + private final Session session; + private final MessageProducer producer; public ActivemqClient(ReadonlyConfig config) { this.config = config; try { + int deliveryMode = config.get(DELIVERY_MODE); + if (deliveryMode != 1 && deliveryMode != 2) { + throw new IllegalArgumentException( + "delivery_mode must be 1 (NON_PERSISTENT) or 2 (PERSISTENT), got: " + + deliveryMode); + } + int priority = config.get(PRIORITY); + if (priority < 0 || priority > 9) { + throw new IllegalArgumentException( + "priority must be between 0 and 9, got: " + priority); + } + this.connectionFactory = getConnectionFactory(); log.info("connection factory created"); this.connection = createConnection(config); + this.connection.start(); + this.session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Destination destination = session.createQueue(config.get(QUEUE_NAME)); + this.producer = session.createProducer(destination); + this.producer.setDeliveryMode(deliveryMode); + this.producer.setTimeToLive(config.get(TIME_TO_LIVE)); + this.producer.setPriority(priority); log.info("connection created"); } catch (Exception e) { log.error("Error while creating AMQ client", e); + // Best-effort cleanup of partially-created resources to avoid leaks + try { + close(); + } catch (Exception closeError) { + e.addSuppressed(closeError); + } throw new ActivemqConnectorException( ActivemqConnectorErrorCode.CREATE_ACTIVEMQ_CLIENT_FAILED, "Error while create AMQ client ", @@ -72,7 +108,9 @@ public class ActivemqClient { } public ActiveMQConnectionFactory getConnectionFactory() { - log.info("broker url : " + config.get(URI)); + // Strip credentials from URI before logging to avoid leaking sensitive information + String safeUri = config.get(URI).replaceAll("(://)[^@:]+:[^@]+@", "$1***@"); + log.info("broker url : " + safeUri); ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(config.get(URI)); if (config.get(ALWAYS_SESSION_ASYNC) != null) { @@ -109,39 +147,66 @@ public class ActivemqClient { if (config.get(NESTED_MAP_AND_LIST_ENABLED) != null) { factory.setNestedMapAndListEnabled(config.get(NESTED_MAP_AND_LIST_ENABLED)); } + + // Options declared with .defaultValue() use getOptional() so the setter is only called + // when the user explicitly set the option. Otherwise the ActiveMQConnectionFactory would + // always apply the SeaTunnel default, silently overwriting jms.* parameters from the URI. + config.getOptional(MAX_THREAD_POOL_SIZE).ifPresent(factory::setMaxThreadPoolSize); + config.getOptional(SEND_TIMEOUT).ifPresent(factory::setSendTimeout); + config.getOptional(USE_COMPRESSION).ifPresent(factory::setUseCompression); + config.getOptional(CONNECT_RESPONSE_TIMEOUT).ifPresent(factory::setConnectResponseTimeout); + config.getOptional(PRODUCER_WINDOW_SIZE).ifPresent(factory::setProducerWindowSize); + config.getOptional(USE_ASYNC_SEND).ifPresent(factory::setUseAsyncSend); 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); + TextMessage textMessage = session.createTextMessage(messageBody); + producer.send(textMessage); } 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)), + String.format("Cannot send AMQ message to queue %s", config.get(QUEUE_NAME)), e); } } public void close() { + JMSException firstError = null; + try { + if (producer != null) { + producer.close(); + } + } catch (JMSException e) { + firstError = e; + } + try { + if (session != null) { + session.close(); + } + } catch (JMSException e) { + if (firstError == null) { + firstError = e; + } + } try { if (connection != null) { connection.close(); } } catch (JMSException e) { + if (firstError == null) { + firstError = e; + } + } + if (firstError != null) { throw new ActivemqConnectorException( ActivemqConnectorErrorCode.CLOSE_CONNECTION_FAILED, String.format( - "Error while closing AMQ connection with %s", config.get(QUEUE_NAME))); + "Error while closing AMQ connection with %s", config.get(QUEUE_NAME)), + firstError); } } diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSinkOptions.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSinkOptions.java index 1d4300b004..7806f84072 100644 --- a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSinkOptions.java +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSinkOptions.java @@ -122,4 +122,81 @@ public class ActivemqSinkOptions implements Serializable { .withDescription( "Controls whether message expiration checking is done in each " + "MessageConsumer prior to dispatching a message."); + + public static final Option<Integer> MAX_THREAD_POOL_SIZE = + Options.key("max_thread_pool_size") + .intType() + .defaultValue(1000) + .withDescription( + "Sets the maximum thread pool size for the ActiveMQ connection factory's " + + "internal thread pool used for dispatching messages."); + + public static final Option<Integer> SEND_TIMEOUT = + Options.key("send_timeout") + .intType() + .defaultValue(0) + .withDescription( + "Sets the send timeout in milliseconds for the ActiveMQ connection factory. " + + "A value of 0 means no timeout."); + + public static final Option<Boolean> USE_COMPRESSION = + Options.key("use_compression") + .booleanType() + .defaultValue(false) + .withDescription( + "When true, message bodies are compressed before being sent to the broker."); + + public static final Option<Integer> CONNECT_RESPONSE_TIMEOUT = + Options.key("connect_response_timeout") + .intType() + .defaultValue(0) + .withDescription( + "Timeout in milliseconds for waiting for the broker's response during " + + "connection establishment. A value of 0 means no timeout. " + + "Defaults to 0."); + + public static final Option<Integer> DELIVERY_MODE = + Options.key("delivery_mode") + .intType() + .defaultValue(2) + .withDescription( + "JMS delivery mode: 1 for NON_PERSISTENT (message kept in memory, " + + "lost if broker crashes) or 2 for PERSISTENT (message written " + + "to disk before acknowledge). Defaults to 2 (PERSISTENT)."); + + public static final Option<Integer> PRIORITY = + Options.key("priority") + .intType() + .defaultValue(4) + .withDescription( + "JMS message priority from 0 (lowest) to 9 (highest). The broker " + + "delivers higher-priority messages before lower-priority ones. " + + "Defaults to 4."); + + public static final Option<Integer> PRODUCER_WINDOW_SIZE = + Options.key("producer_window_size") + .intType() + .defaultValue(0) + .withDescription( + "Flow control window size in bytes for async sends. A value of 0 means " + + "no flow control. Used together with use_async_send to prevent " + + "unbounded memory growth. Defaults to 0."); + + public static final Option<Long> TIME_TO_LIVE = + Options.key("time_to_live") + .longType() + .defaultValue(0L) + .withDescription( + "Message time-to-live in milliseconds. A value of 0 means the message " + + "never expires. The broker discards messages after the TTL " + + "expires. Defaults to 0."); + + public static final Option<Boolean> USE_ASYNC_SEND = + Options.key("use_async_send") + .booleanType() + .defaultValue(false) + .withDescription( + "When true, the producer sends messages without waiting for the broker " + + "to confirm receipt, improving throughput at the risk of message " + + "loss if the broker fails. Defaults to false."); } diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/sink/ActivemqSinkFactory.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/sink/ActivemqSinkFactory.java index 55d14771a1..d7ec4aea58 100644 --- a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/sink/ActivemqSinkFactory.java +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/sink/ActivemqSinkFactory.java @@ -31,13 +31,22 @@ import static org.apache.seatunnel.connectors.seatunnel.activemq.config.Activemq import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CHECK_FOR_DUPLICATE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CLIENT_ID; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CLOSE_TIMEOUT; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CONNECT_RESPONSE_TIMEOUT; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.CONSUMER_EXPIRY_CHECK_ENABLED; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.DELIVERY_MODE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.DISPATCH_ASYNC; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.MAX_THREAD_POOL_SIZE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.NESTED_MAP_AND_LIST_ENABLED; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PASSWORD; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PRIORITY; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PRODUCER_WINDOW_SIZE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.QUEUE_NAME; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.SEND_TIMEOUT; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.TIME_TO_LIVE; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.URI; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.USERNAME; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.USE_ASYNC_SEND; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.USE_COMPRESSION; import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT; @AutoService(Factory.class) @@ -63,7 +72,16 @@ public class ActivemqSinkFactory implements TableSinkFactory { CONSUMER_EXPIRY_CHECK_ENABLED, DISPATCH_ASYNC, NESTED_MAP_AND_LIST_ENABLED, - WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT) + WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT, + MAX_THREAD_POOL_SIZE, + SEND_TIMEOUT, + USE_COMPRESSION, + CONNECT_RESPONSE_TIMEOUT, + DELIVERY_MODE, + PRIORITY, + PRODUCER_WINDOW_SIZE, + TIME_TO_LIVE, + USE_ASYNC_SEND) .build(); } diff --git a/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/ActivemqFactoryTest.java b/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/ActivemqFactoryTest.java index 2141330010..4517eb055b 100644 --- a/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/ActivemqFactoryTest.java +++ b/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/ActivemqFactoryTest.java @@ -90,6 +90,15 @@ class ActivemqFactoryTest { config.put(ActivemqSinkOptions.CLOSE_TIMEOUT.key(), 1000); config.put(ActivemqSinkOptions.CONSUMER_EXPIRY_CHECK_ENABLED.key(), true); config.put(ActivemqSinkOptions.WARN_ABOUT_UNSTARTED_CONNECTION_TIMEOUT.key(), -1); + config.put(ActivemqSinkOptions.MAX_THREAD_POOL_SIZE.key(), 100); + config.put(ActivemqSinkOptions.SEND_TIMEOUT.key(), 5000); + config.put(ActivemqSinkOptions.USE_COMPRESSION.key(), true); + config.put(ActivemqSinkOptions.CONNECT_RESPONSE_TIMEOUT.key(), 5000); + config.put(ActivemqSinkOptions.DELIVERY_MODE.key(), 1); + config.put(ActivemqSinkOptions.PRIORITY.key(), 9); + config.put(ActivemqSinkOptions.PRODUCER_WINDOW_SIZE.key(), 1048576); + config.put(ActivemqSinkOptions.TIME_TO_LIVE.key(), 60000); + config.put(ActivemqSinkOptions.USE_ASYNC_SEND.key(), true); Assertions.assertDoesNotThrow(() -> validate(config)); } diff --git a/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClientTest.java b/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClientTest.java new file mode 100644 index 0000000000..c2f9bfcb94 --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClientTest.java @@ -0,0 +1,510 @@ +/* + * 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.config.ActivemqSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.activemq.exception.ActivemqConnectorException; + +import org.apache.activemq.ActiveMQConnectionFactory; + +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; +import org.mockito.MockedConstruction; + +import javax.jms.Connection; +import javax.jms.JMSException; +import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.TextMessage; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.anyBoolean; +import static org.mockito.Mockito.anyInt; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ActivemqClientTest { + + private static final String TEST_URI = "tcp://localhost:61616"; + private static final String TEST_QUEUE = "test-queue"; + private static final String TEST_URI_WITH_CREDENTIALS = + "tcp://admin:secretPass@localhost:61616"; + + private static Map<String, Object> baseConfig() { + Map<String, Object> config = new HashMap<>(); + config.put(ActivemqSinkOptions.URI.key(), TEST_URI); + config.put(ActivemqSinkOptions.QUEUE_NAME.key(), TEST_QUEUE); + return config; + } + + /** + * L1 + L2: Session and MessageProducer are created once in the constructor, and + * connection.start() is called once in the constructor. + */ + @Test + void constructorCreatesSessionProducerAndStartsConnection() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + + new ActivemqClient(ReadonlyConfig.fromMap(baseConfig())); + + // L2: connection.start() called once in constructor + verify(mockConnection).start(); + // L1: session created once in constructor + verify(mockConnection).createSession(false, Session.AUTO_ACKNOWLEDGE); + // L1: producer created once in constructor + verify(mockSession).createProducer(mockQueue); + } + } + + /** + * L1: write() reuses the session and producer created in the constructor instead of creating + * new JMS resources for each message. + */ + @Test + void writeReusesSessionAndProducer() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + TextMessage mockTextMessage = mock(TextMessage.class); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + when(mockSession.createTextMessage("hello")).thenReturn(mockTextMessage); + + ActivemqClient client = new ActivemqClient(ReadonlyConfig.fromMap(baseConfig())); + client.write("hello".getBytes(StandardCharsets.UTF_8)); + + // L1: createSession still called only once (constructor only, not in write) + verify(mockConnection, times(1)).createSession(false, Session.AUTO_ACKNOWLEDGE); + // L1: createProducer still called only once (constructor only, not in write) + verify(mockSession, times(1)).createProducer(mockQueue); + // L2: connection.start() not called again in write() + verify(mockConnection, times(1)).start(); + // write() creates a TextMessage and sends it via the reused producer + verify(mockSession).createTextMessage("hello"); + verify(mockProducer).send(mockTextMessage); + } + } + + /** + * L4 + exception chain: write() wraps JMSException with the queue name (not the broker URI, + * which may contain credentials) in the error message, and preserves the original exception as + * the cause. + */ + @Test + void writeWrapsJmsExceptionWithQueueInMessage() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + JMSException sendError = new JMSException("send failed"); + + Map<String, Object> config = baseConfig(); + config.put(ActivemqSinkOptions.URI.key(), TEST_URI_WITH_CREDENTIALS); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + when(mockSession.createTextMessage("hello")).thenThrow(sendError); + + ActivemqClient client = new ActivemqClient(ReadonlyConfig.fromMap(config)); + + ActivemqConnectorException ex = + assertThrows( + ActivemqConnectorException.class, + () -> client.write("hello".getBytes(StandardCharsets.UTF_8))); + + // Issue 2: error message contains queue name, not the broker URI + assertTrue(ex.getMessage().contains(TEST_QUEUE)); + // Issue 2: credentials from the URI must not appear in the message + assertFalse(ex.getMessage().contains("secretPass")); + assertFalse(ex.getMessage().contains(TEST_URI_WITH_CREDENTIALS)); + // Exception chain preserved (3-arg constructor) + assertEquals(sendError, ex.getCause()); + } + } + + /** L1: close() closes Producer, Session, and Connection in reverse creation order. */ + @Test + void closeClosesAllResourcesInReverseOrder() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + + ActivemqClient client = new ActivemqClient(ReadonlyConfig.fromMap(baseConfig())); + client.close(); + + InOrder order = inOrder(mockProducer, mockSession, mockConnection); + order.verify(mockProducer).close(); + order.verify(mockSession).close(); + order.verify(mockConnection).close(); + } + } + + /** + * L3: close() preserves the first JMSException as the cause and still closes all remaining + * resources (session and connection) even when an earlier close fails. + */ + @Test + void closePreservesFirstExceptionAndClosesRemainingResources() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + JMSException producerError = new JMSException("producer close failed"); + JMSException sessionError = new JMSException("session close failed"); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + doThrow(producerError).when(mockProducer).close(); + doThrow(sessionError).when(mockSession).close(); + + ActivemqClient client = new ActivemqClient(ReadonlyConfig.fromMap(baseConfig())); + + ActivemqConnectorException ex = + assertThrows(ActivemqConnectorException.class, () -> client.close()); + + // L3: first error (producer) is preserved as the cause + assertEquals(producerError, ex.getCause()); + // All resources still closed despite errors + verify(mockProducer).close(); + verify(mockSession).close(); + verify(mockConnection).close(); + } + } + + /** + * L3: close() wraps a connection-only close failure with the original JMSException as the + * cause. + */ + @Test + void closeWithConnectionErrorOnlyWrapsCause() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + JMSException connError = new JMSException("connection close failed"); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + doThrow(connError).when(mockConnection).close(); + + ActivemqClient client = new ActivemqClient(ReadonlyConfig.fromMap(baseConfig())); + + ActivemqConnectorException ex = + assertThrows(ActivemqConnectorException.class, () -> client.close()); + + // L3: connection close error is preserved as the cause + assertEquals(connError, ex.getCause()); + verify(mockProducer).close(); + verify(mockSession).close(); + verify(mockConnection).close(); + } + } + + /** + * When factory-level options are not explicitly set, their setters must NOT be called so that + * {@code jms.*} parameters embedded in the broker URI are preserved. Producer-level options + * (delivery_mode, time_to_live, priority) always apply their JMS defaults. + */ + @Test + void doesNotCallFactorySettersWhenOptionsNotSet() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + + new ActivemqClient(ReadonlyConfig.fromMap(baseConfig())); + + ActiveMQConnectionFactory factory = ignored.constructed().get(0); + // Issue 1: factory setters must NOT be called when options use defaults + verify(factory, times(0)).setMaxThreadPoolSize(anyInt()); + verify(factory, times(0)).setSendTimeout(anyInt()); + verify(factory, times(0)).setUseCompression(anyBoolean()); + verify(factory, times(0)).setConnectResponseTimeout(anyInt()); + verify(factory, times(0)).setProducerWindowSize(anyInt()); + verify(factory, times(0)).setUseAsyncSend(anyBoolean()); + // Producer defaults are always applied (they default to JMS defaults) + verify(mockProducer).setDeliveryMode(2); + verify(mockProducer).setTimeToLive(0L); + verify(mockProducer).setPriority(4); + } + } + + /** + * When the broker URI contains {@code jms.*} query parameters and the user does not set the + * corresponding SeaTunnel options, the factory setters must NOT be called so that the URI + * parameters are preserved. This test uses {@code mockConstruction} to verify that the + * default-value setters are never invoked; it does not exercise a real factory. + */ + @Test + void preservesUriJmsParamsWhenOptionsNotSet() throws Exception { + String uriWithJmsParams = + "tcp://localhost:61616?jms.useAsyncSend=true&jms.producerWindowSize=1048576"; + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + + Map<String, Object> config = baseConfig(); + config.put(ActivemqSinkOptions.URI.key(), uriWithJmsParams); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + + new ActivemqClient(ReadonlyConfig.fromMap(config)); + + ActiveMQConnectionFactory factory = ignored.constructed().get(0); + // Issue 1: URI jms.* params must not be overwritten by SeaTunnel defaults + verify(factory, times(0)).setUseAsyncSend(false); + verify(factory, times(0)).setProducerWindowSize(0); + } + } + + /** + * Verifies that explicitly configured properties are applied to the connection factory and the + * producer during construction. + */ + @Test + void appliesOptionalProperties() throws Exception { + Connection mockConnection = mock(Connection.class); + Session mockSession = mock(Session.class); + Queue mockQueue = mock(Queue.class); + MessageProducer mockProducer = mock(MessageProducer.class); + + Map<String, Object> config = baseConfig(); + config.put(ActivemqSinkOptions.MAX_THREAD_POOL_SIZE.key(), 100); + config.put(ActivemqSinkOptions.SEND_TIMEOUT.key(), 5000); + config.put(ActivemqSinkOptions.USE_COMPRESSION.key(), true); + config.put(ActivemqSinkOptions.CONNECT_RESPONSE_TIMEOUT.key(), 5000); + config.put(ActivemqSinkOptions.PRODUCER_WINDOW_SIZE.key(), 1048576); + config.put(ActivemqSinkOptions.USE_ASYNC_SEND.key(), true); + config.put(ActivemqSinkOptions.DELIVERY_MODE.key(), 1); + config.put(ActivemqSinkOptions.TIME_TO_LIVE.key(), 60000L); + config.put(ActivemqSinkOptions.PRIORITY.key(), 9); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenReturn(mockSession); + when(mockSession.createQueue(TEST_QUEUE)).thenReturn(mockQueue); + when(mockSession.createProducer(mockQueue)).thenReturn(mockProducer); + + new ActivemqClient(ReadonlyConfig.fromMap(config)); + + ActiveMQConnectionFactory factory = ignored.constructed().get(0); + // ConnectionFactory explicit values + verify(factory).setMaxThreadPoolSize(100); + verify(factory).setSendTimeout(5000); + verify(factory).setUseCompression(true); + verify(factory).setConnectResponseTimeout(5000); + verify(factory).setProducerWindowSize(1048576); + verify(factory).setUseAsyncSend(true); + // Producer explicit values + verify(mockProducer).setDeliveryMode(1); + verify(mockProducer).setTimeToLive(60000L); + verify(mockProducer).setPriority(9); + } + } + + /** + * P1: When the constructor fails partway through, partially-created resources (e.g. connection) + * are cleaned up via close() to avoid leaks, and the original construction exception is + * propagated as the cause. + */ + @Test + void constructorFailureCleansUpPartiallyCreatedResources() throws Exception { + Connection mockConnection = mock(Connection.class); + JMSException constructionError = new JMSException("createSession failed"); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + // connection.start() succeeds, then createSession fails + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenThrow(constructionError); + + ActivemqConnectorException ex = + assertThrows( + ActivemqConnectorException.class, + () -> new ActivemqClient(ReadonlyConfig.fromMap(baseConfig()))); + + // Original construction error is preserved as the cause + assertEquals(constructionError, ex.getCause()); + // P1: connection was cleaned up despite construction failure + verify(mockConnection).close(); + } + } + + /** + * Issue 5: When the constructor fails and best-effort close() also throws, the close error is + * attached as a suppressed exception on the original cause instead of being swallowed. + */ + @Test + void constructorFailurePreservesCloseErrorAsSuppressed() throws Exception { + Connection mockConnection = mock(Connection.class); + JMSException constructionError = new JMSException("createSession failed"); + JMSException closeError = new JMSException("connection close failed"); + + try (MockedConstruction<ActiveMQConnectionFactory> ignored = + mockConstruction( + ActiveMQConnectionFactory.class, + (factory, ctx) -> + when(factory.createConnection()).thenReturn(mockConnection))) { + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)) + .thenThrow(constructionError); + doThrow(closeError).when(mockConnection).close(); + + ActivemqConnectorException ex = + assertThrows( + ActivemqConnectorException.class, + () -> new ActivemqClient(ReadonlyConfig.fromMap(baseConfig()))); + + // Original construction error is preserved as the cause + assertEquals(constructionError, ex.getCause()); + // Issue 5: close() wraps JMSException in ActivemqConnectorException, which is then + // attached as a suppressed exception instead of being swallowed + assertEquals(1, ex.getCause().getSuppressed().length); + assertTrue(ex.getCause().getSuppressed()[0] instanceof ActivemqConnectorException); + } + } + + /** + * Issue 3: delivery_mode must be 1 (NON_PERSISTENT) or 2 (PERSISTENT). An invalid value should + * fail fast with a clear message instead of surfacing as a generic construction error. + */ + @Test + void constructorRejectsInvalidDeliveryMode() { + Map<String, Object> config = baseConfig(); + config.put(ActivemqSinkOptions.DELIVERY_MODE.key(), 5); + + ActivemqConnectorException ex = + assertThrows( + ActivemqConnectorException.class, + () -> new ActivemqClient(ReadonlyConfig.fromMap(config))); + + assertTrue(ex.getCause() instanceof IllegalArgumentException); + assertTrue(ex.getCause().getMessage().contains("delivery_mode")); + } + + /** + * Issue 3: priority must be between 0 and 9. An out-of-range value should fail fast with a + * clear message. + */ + @Test + void constructorRejectsInvalidPriority() { + Map<String, Object> config = baseConfig(); + config.put(ActivemqSinkOptions.PRIORITY.key(), 15); + + ActivemqConnectorException ex = + assertThrows( + ActivemqConnectorException.class, + () -> new ActivemqClient(ReadonlyConfig.fromMap(config))); + + assertTrue(ex.getCause() instanceof IllegalArgumentException); + assertTrue(ex.getCause().getMessage().contains("priority")); + } +}
