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-12341-7e715b039f0c33ff002fbf6a56a7a5d6522ab167 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c6ddef67e4f7e799e012ea2bf73ab9cc869f82f9 Author: Goutam Adwant <[email protected]> AuthorDate: Fri Oct 9 09:25:44 2026 +0000 [Feature][Connector-V2] Add checkpoint-aware ActiveMQ queue source (#12341) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/connectors/changelog/connector-activemq.md | 1 + docs/en/connectors/sink/Activemq.md | 5 +- docs/en/connectors/source/Activemq.md | 129 +++++ docs/zh/connectors/changelog/connector-activemq.md | 7 + docs/zh/connectors/sink/Activemq.md | 5 +- docs/zh/connectors/source/Activemq.md | 129 +++++ plugin-mapping.properties | 1 + .../activemq/config/ActivemqSourceOptions.java | 51 ++ .../seatunnel/activemq/source/ActivemqSource.java | 98 ++++ .../activemq/source/ActivemqSourceEnumerator.java | 137 +++++ .../activemq/source/ActivemqSourceFactory.java | 131 +++++ .../activemq/source/ActivemqSourceReader.java | 366 ++++++++++++++ .../seatunnel/activemq/ActivemqFactoryTest.java | 97 ++++ .../activemq/source/ActivemqSourceReaderTest.java | 553 +++++++++++++++++++++ .../e2e/connector/activemq/ActivemqIT.java | 284 ++++++++++- .../test/resources/activemq_source_to_console.conf | 39 ++ 16 files changed, 2018 insertions(+), 15 deletions(-) diff --git a/docs/en/connectors/changelog/connector-activemq.md b/docs/en/connectors/changelog/connector-activemq.md index 4a9eb97933..f8d1157bb0 100644 --- a/docs/en/connectors/changelog/connector-activemq.md +++ b/docs/en/connectors/changelog/connector-activemq.md @@ -2,6 +2,7 @@ | Change | Commit | Version | | --- | --- | --- | +|[Feature][Connector-V2] Add checkpoint-aware ActiveMQ queue source|-|Next| |[Improve][connector][activemq] Remove duplicate dependencies (#8753)|https://github.com/apache/seatunnel/commit/da6241aa1c|2.3.10| |[improve] update activemq connector config option (#8580)|https://github.com/apache/seatunnel/commit/629f85b23a|2.3.10| |[Improve][dist]add shade check rule (#8136)|https://github.com/apache/seatunnel/commit/51ef800016|2.3.9| diff --git a/docs/en/connectors/sink/Activemq.md b/docs/en/connectors/sink/Activemq.md index e5590e5f5c..c3aa17156c 100644 --- a/docs/en/connectors/sink/Activemq.md +++ b/docs/en/connectors/sink/Activemq.md @@ -12,8 +12,8 @@ import ChangeLog from '../changelog/connector-activemq.md'; ## Description -Write SeaTunnel rows to an ActiveMQ queue. Each row is serialized as a JSON text message. This is -a sink-only connector; SeaTunnel does not provide an ActiveMQ source connector. +Write SeaTunnel rows to an ActiveMQ queue. Each row is serialized as a JSON text message. +For consuming queues, see the separate [ActiveMQ source](../source/Activemq.md). ## Key features @@ -135,4 +135,3 @@ No. The sink is best-effort with bounded reconnect behavior driven by the underl ## Changelog <ChangeLog /> - diff --git a/docs/en/connectors/source/Activemq.md b/docs/en/connectors/source/Activemq.md new file mode 100644 index 0000000000..47698a7ec1 --- /dev/null +++ b/docs/en/connectors/source/Activemq.md @@ -0,0 +1,129 @@ +import ChangeLog from '../changelog/connector-activemq.md'; + +# ActiveMQ + +<!-- +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. +--> + +> ActiveMQ Classic queue source connector + +## Description + +Reads JMS `TextMessage` payloads from one ActiveMQ Classic queue over OpenWire. Reuses the existing ActiveMQ module and JSON/text deserializers. This is not a generic JMS or AMQP connector. + +## Support Those Engines + +> Flink<br/> +> SeaTunnel Zeta<br/> + +The source requires streaming mode and periodic checkpoints. Spark is not supported by this first slice: its virtual source checkpoints have not been validated as an end-to-end commit boundary for destructive queue acknowledgements. + +## Key Features + +- [ ] [batch](../../introduction/concepts/connector-v2-features.md) +- [x] [stream](../../introduction/concepts/connector-v2-features.md) +- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) +- [x] [parallelism](../../introduction/concepts/connector-v2-features.md) +- [ ] [column projection](../../introduction/concepts/connector-v2-features.md) +- [ ] [support user-defined split](../../introduction/concepts/connector-v2-features.md) + +## Options + +| Name | Type | Required | Default | +| --- | --- | --- | --- | +| uri | string | yes | - | +| queue_name | string | yes | - | +| username | string | no | - | +| password | string | no | - | +| format | enum | no | JSON | +| field_delimiter | string | no | , | +| max_in_flight_messages | int | no | 1000 | +| schema | config | yes | - | + +### uri [string] + +One broker endpoint: `tcp://host:port` or `ssl://host:port`. Credentials, URI query options, composite transports (including `failover:`), and URI paths are not supported. Use the separate credential options. For TLS, configure the JVM trust/key stores on every worker and use `ssl://`; never disable certificate validation. Plain TCP should only be used on a trusted network. + +The connector fails the task on a broken connection so the engine restores the reader with a new connection. It does not reconnect in place or reuse acknowledgement handles from an earlier session. + +### queue_name [string] + +One literal queue name. Wildcards, composite queues, destination prefixes and `?consumer.*` destination options are rejected. Topics, durable topic subscriptions and selectors are outside this first slice. Provision the queue and the broker account's permissions before starting the job; broker auto-creation policy is not controlled by the connector. + +### username / password [string] + +Optional credentials, which must be configured together. When omitted, the client connects without explicit credentials. Use a broker account scoped to the intended queue. Do not embed credentials in `uri`. + +### format / field_delimiter + +`JSON` maps an object or an array of objects using `schema`. `TEXT` uses the existing delimited text deserializer, with `field_delimiter` (default `,`). Only JMS `TextMessage` is accepted; `ObjectMessage`, `BytesMessage`, null or empty bodies, and invalid payloads fail the task without acknowledging that message. No Java object payload is deserialized. Deserialization/emission failures omit the original exception details because those details can contain the queue payload; verify the sch [...] + +### max_in_flight_messages [int] + +Positive maximum count of emitted, unacknowledged messages per reader. The source pauses receiving when this limit is reached and resumes after checkpoint completion. The ActiveMQ client queue prefetch is set to the same count; client-side prefetched messages may consume additional memory. This is not a byte limit. Size the limit and producer message sizes for available worker memory. Frequent successful checkpoints are needed for sustained throughput. + +### schema [config] + +Payload schema; see [Schema Feature](../../introduction/concepts/schema-feature.md). Message headers/properties are not exposed as metadata in this slice. + +### Common Options + +See [Source Common Options](../common-options/source-common-options.md). + +## Delivery and Recovery + +- Delivery is **at least once**, not exactly once. Each message is individually acknowledged only after a checkpoint containing its emitted rows completes. A later completed checkpoint can include messages from earlier aborted checkpoints. +- Failure before acknowledgement closes the connection and leaves messages eligible for broker redelivery. Duplicates are possible after recovery, including partially emitted multi-row JSON messages; use an idempotent downstream sink where needed. +- Checkpoint state contains logical consumer slots, not broker message objects or queue offsets. ActiveMQ remains responsible for retaining and redelivering unacknowledged messages. +- Each active parallel reader owns a connection, session and competing queue consumer. Normal startup assigns one slot per reader. Recovery preserves existing logical slots and adds slots when parallelism increases; a reader can own multiple restored slots while using one consumer after scale-down. No global ordering is guaranteed with parallel consumers or redelivery. +- An empty queue is not end-of-input. The source continues waiting until the job is cancelled or fails. +- Retention depends on broker persistence, producer delivery mode, expiration and dead-letter/redelivery policies. A durable broker and persistent messages are required to survive broker restarts. A permanently invalid message can repeatedly fail the job or be routed to a dead-letter queue by broker policy; this connector never silently skips it. + +## Example + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + ActiveMQ { + uri = "tcp://activemq-host:61616" + queue_name = "events" + format = JSON + max_in_flight_messages = 1000 + schema { + fields { + id = bigint + name = string + } + } + } +} + +sink { + Console {} +} +``` + +For delimited messages such as `42|created`, set `format = TEXT` and `field_delimiter = "|"`, with schema fields in the payload's field order. Existing ActiveMQ **sink** configuration and behavior are unchanged; source options do not apply to the sink. + +## Changelog + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-activemq.md b/docs/zh/connectors/changelog/connector-activemq.md index e69de29bb2..c9452522e5 100644 --- a/docs/zh/connectors/changelog/connector-activemq.md +++ b/docs/zh/connectors/changelog/connector-activemq.md @@ -0,0 +1,7 @@ +<details><summary> 变更日志 </summary> + +| 变更 | 提交 | 版本 | +| --- | --- | --- | +|[Feature][Connector-V2] Add checkpoint-aware ActiveMQ queue source|-|Next| + +</details> diff --git a/docs/zh/connectors/sink/Activemq.md b/docs/zh/connectors/sink/Activemq.md index c9f045ce25..7cba3be4e3 100644 --- a/docs/zh/connectors/sink/Activemq.md +++ b/docs/zh/connectors/sink/Activemq.md @@ -6,8 +6,8 @@ import ChangeLog from '../changelog/connector-activemq.md'; ## 描述 -用于把 SeaTunnel 数据写入 ActiveMQ 队列。每一行数据都会被序列化成一条 JSON 文本消息。这个连接器只支持 -Sink，SeaTunnel 目前没有提供 ActiveMQ Source 连接器。 +用于把 SeaTunnel 数据写入 ActiveMQ 队列。每一行数据都会被序列化成一条 JSON 文本消息。 +消费队列请参见独立的 [ActiveMQ Source](../source/Activemq.md)。 ## 关键特性 @@ -128,4 +128,3 @@ sink { ## 变更日志 <ChangeLog /> - diff --git a/docs/zh/connectors/source/Activemq.md b/docs/zh/connectors/source/Activemq.md new file mode 100644 index 0000000000..123aba5dd7 --- /dev/null +++ b/docs/zh/connectors/source/Activemq.md @@ -0,0 +1,129 @@ +import ChangeLog from '../changelog/connector-activemq.md'; + +# ActiveMQ + +<!-- +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. +--> + +> ActiveMQ Classic 队列数据源连接器 + +## 描述 + +通过 OpenWire 从一个 ActiveMQ Classic 队列读取 JMS `TextMessage`，复用现有 ActiveMQ 模块以及 JSON 和文本反序列化器。本连接器不是通用 JMS 或 AMQP 实现。 + +## 支持的引擎 + +> Flink<br/> +> SeaTunnel Zeta<br/> + +数据源要求流模式和周期性检查点。本次不支持 Spark：尚未验证其虚拟数据源检查点是否能作为删除队列消息所需的端到端提交边界。 + +## 主要特性 + +- [ ] [批处理](../../introduction/concepts/connector-v2-features.md) +- [x] [流处理](../../introduction/concepts/connector-v2-features.md) +- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) +- [x] [并行度](../../introduction/concepts/connector-v2-features.md) +- [ ] [列投影](../../introduction/concepts/connector-v2-features.md) +- [ ] [自定义分片](../../introduction/concepts/connector-v2-features.md) + +## 配置项 + +| 名称 | 类型 | 必填 | 默认值 | +| --- | --- | --- | --- | +| uri | string | 是 | - | +| queue_name | string | 是 | - | +| username | string | 否 | - | +| password | string | 否 | - | +| format | enum | 否 | JSON | +| field_delimiter | string | 否 | , | +| max_in_flight_messages | int | 否 | 1000 | +| schema | config | 是 | - | + +### uri [string] + +单个代理地址，格式为 `tcp://host:port` 或 `ssl://host:port`。不支持 URI 中的凭据、查询参数、路径或包括 `failover:` 在内的复合传输。凭据必须使用单独的配置项。TLS 连接应在每个工作节点配置 JVM 信任库和密钥库并使用 `ssl://`，不要禁用证书校验。明文 TCP 仅应在可信网络内使用。 + +连接中断时任务失败，由引擎恢复读取器并创建新连接，不在原会话内重连或复用旧确认句柄。 + +### queue_name [string] + +一个普通队列名称，不支持通配符、复合队列、目标前缀或 `?consumer.*` 参数。本次不支持主题、持久主题订阅和选择器。应提前创建队列并配置账号权限；连接器不控制代理的自动创建策略。 + +### username / password [string] + +可选凭据，必须同时配置。省略时客户端不提供显式凭据。账号权限应限定到目标队列，不要在 URI 中嵌入凭据。 + +### format / field_delimiter + +`JSON` 按 `schema` 映射对象或对象数组；`TEXT` 使用现有分隔文本反序列化器，分隔符 `field_delimiter` 默认为 `,`。仅支持 `TextMessage`；对象消息、字节消息、null 或空字符串消息体以及无效内容使任务失败且不确认该消息，不进行 Java 对象反序列化。反序列化或输出失败时不保留可能包含消息内容的原始异常详情；请通过有权限的代理客户端检查消息与 schema、format 是否匹配。 + +### max_in_flight_messages [int] + +每个读取器允许保留的已输出但尚未确认的消息数，必须大于零。达到上限后暂停接收，完成检查点后恢复。客户端队列预取数设置为相同值，预取消息可能额外占用内存。这是数量限制而不是字节限制，应按工作节点内存和生产者消息大小配置。持续吞吐依赖定期成功完成的检查点。 + +### schema [config] + +消息内容的模式，参见[模式特性](../../introduction/concepts/schema-feature.md)。本次不将消息头或属性暴露为元数据。 + +### 公共配置 + +参见[数据源公共配置](../common-options/source-common-options.md)。 + +## 交付与恢复 + +- 语义为**至少一次**，不是精确一次。只有包含已输出行的检查点完成后，才逐条确认消息。后续成功检查点也可覆盖之前中止的检查点中的消息。 +- 确认前失败会关闭连接，未确认消息可由代理重新投递。恢复可能产生重复行，包括部分输出的多行 JSON 消息；需要时使用幂等下游。 +- 检查点只保存逻辑消费者分片，不保存 JMS 消息对象或队列偏移量；代理负责保留和重新投递未确认消息。 +- 每个活跃并行读取器拥有独立连接、会话和竞争消费者。初次启动每个读取器分配一个分片；恢复保留已有逻辑分片，增加并行度时补充新分片。缩容后一个读取器可以持有多个恢复分片，但只使用一个消费者。并行消费和重投递不保证全局顺序。 +- 空队列不表示输入结束，读取器一直等待到作业取消或失败。 +- 保留和重投递受代理持久化、消息交付模式、过期时间及死信策略影响。要跨代理重启保留消息，需要持久代理及持久消息。永久无效消息可能重复导致任务失败或按代理策略进入死信队列；连接器不会静默跳过。 + +## 示例 + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + ActiveMQ { + uri = "tcp://activemq-host:61616" + queue_name = "events" + format = JSON + max_in_flight_messages = 1000 + schema { + fields { + id = bigint + name = string + } + } + } +} + +sink { + Console {} +} +``` + +对于 `42|created` 等分隔文本，配置 `format = TEXT`、`field_delimiter = "|"`，并按字段顺序定义模式。现有 ActiveMQ Sink 的配置和行为保持不变；上述 Source 配置不适用于 Sink。 + +## 变更日志 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index 2addbea9f5..643774423d 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -174,6 +174,7 @@ seatunnel.source.BosFile = connector-file-bos seatunnel.sink.BosFile = connector-file-bos seatunnel.source.Milvus = connector-milvus seatunnel.sink.Milvus = connector-milvus +seatunnel.source.ActiveMQ = connector-activemq seatunnel.sink.ActiveMQ = connector-activemq seatunnel.source.MQTT = connector-mqtt seatunnel.sink.MQTT = connector-mqtt diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSourceOptions.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSourceOptions.java new file mode 100644 index 0000000000..5806db09ba --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/config/ActivemqSourceOptions.java @@ -0,0 +1,51 @@ +/* + * 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.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; + +/** Options for consuming one ActiveMQ Classic queue. */ +public final class ActivemqSourceOptions { + public enum MessageFormat { + JSON, + TEXT + } + + public static final Option<MessageFormat> FORMAT = + Options.key("format") + .enumType(MessageFormat.class) + .defaultValue(MessageFormat.JSON) + .withDescription("TextMessage payload format: JSON or delimited TEXT."); + + public static final Option<String> FIELD_DELIMITER = + Options.key("field_delimiter") + .stringType() + .defaultValue(",") + .withDescription("Field delimiter for TEXT payloads."); + + public static final Option<Integer> MAX_IN_FLIGHT_MESSAGES = + Options.key("max_in_flight_messages") + .intType() + .defaultValue(1000) + .withDescription( + "Maximum emitted messages retained per reader until checkpoint completion. " + + "Also bounds the client queue prefetch; this is a count, not a byte limit."); + + private ActivemqSourceOptions() {} +} diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSource.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSource.java new file mode 100644 index 0000000000..5829ac5e80 --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSource.java @@ -0,0 +1,98 @@ +/* + * 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.source; + +import org.apache.seatunnel.api.common.JobContext; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.source.Boundedness; +import org.apache.seatunnel.api.source.SeaTunnelSource; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.api.source.SupportParallelism; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.constants.JobMode; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceEnumerator.Split; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceEnumerator.State; + +import java.util.Collections; +import java.util.List; + +/** Competing consumers of one queue; the broker retains messages until checkpoint completion. */ +public class ActivemqSource + implements SeaTunnelSource<SeaTunnelRow, Split, State>, SupportParallelism { + private final ReadonlyConfig config; + private final CatalogTable table; + private final DeserializationSchema<SeaTunnelRow> deserializer; + private JobContext jobContext; + + public ActivemqSource( + ReadonlyConfig config, + CatalogTable table, + DeserializationSchema<SeaTunnelRow> deserializer) { + ActivemqSourceFactory.validate(config); + this.config = config; + this.table = table; + this.deserializer = deserializer; + } + + @Override + public String getPluginName() { + return "ActiveMQ"; + } + + @Override + public Boundedness getBoundedness() { + if (jobContext != null + && (jobContext.getJobMode() != JobMode.STREAMING + || !jobContext.isEnableCheckpoint())) { + throw new IllegalArgumentException( + "ActiveMQ source requires a streaming job with checkpointing enabled"); + } + return Boundedness.UNBOUNDED; + } + + @Override + public void setJobContext(JobContext jobContext) { + this.jobContext = jobContext; + } + + @Override + public List<CatalogTable> getProducedCatalogTables() { + return Collections.singletonList(table); + } + + @Override + public SourceReader<SeaTunnelRow, Split> createReader(SourceReader.Context context) { + getBoundedness(); + return new ActivemqSourceReader(config, deserializer); + } + + @Override + public SourceSplitEnumerator<Split, State> createEnumerator( + SourceSplitEnumerator.Context<Split> context) { + return new ActivemqSourceEnumerator(context); + } + + @Override + public SourceSplitEnumerator<Split, State> restoreEnumerator( + SourceSplitEnumerator.Context<Split> context, State state) { + return new ActivemqSourceEnumerator(context, state); + } +} diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceEnumerator.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceEnumerator.java new file mode 100644 index 0000000000..dd86ee19b6 --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceEnumerator.java @@ -0,0 +1,137 @@ +/* + * 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.source; + +import org.apache.seatunnel.api.source.SourceSplit; +import org.apache.seatunnel.api.source.SourceSplitEnumerator; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** Assigns logical consumer slots, not queue offsets; broker acknowledgements determine replay. */ +public class ActivemqSourceEnumerator + implements SourceSplitEnumerator< + ActivemqSourceEnumerator.Split, ActivemqSourceEnumerator.State> { + private final Context<Split> context; + private final Map<Integer, Split> pending = new LinkedHashMap<>(); + private int consumerSlots; + + public ActivemqSourceEnumerator(Context<Split> context) { + this.context = context; + } + + public ActivemqSourceEnumerator(Context<Split> context, State state) { + this.context = context; + this.consumerSlots = state.consumerSlots; + state.pending.forEach(split -> pending.put(split.id, split)); + } + + @Override + public void open() {} + + /** Preserve restored ownership, adding only new consumer slots when parallelism increases. */ + @Override + public synchronized void run() { + for (int i = consumerSlots; i < context.currentParallelism(); i++) { + pending.putIfAbsent(i, new Split(i)); + } + consumerSlots = Math.max(consumerSlots, context.currentParallelism()); + for (Integer reader : context.registeredReaders()) { + assign(reader); + } + } + + private void assign(int reader) { + if (consumerSlots == 0) { + return; + } + List<Split> splits = new ArrayList<>(); + for (Split split : pending.values()) { + if (split.id % context.currentParallelism() == reader) { + splits.add(split); + } + } + if (!splits.isEmpty()) { + context.assignSplit(reader, splits); + splits.forEach(split -> pending.remove(split.id)); + } + context.signalNoMoreSplits(reader); + } + + @Override + public synchronized void addSplitsBack(List<Split> splits, int subtaskId) { + splits.forEach(split -> pending.put(split.id, split)); + for (Integer reader : context.registeredReaders()) { + assign(reader); + } + } + + @Override + public synchronized void registerReader(int subtaskId) { + assign(subtaskId); + } + + @Override + public synchronized void handleSplitRequest(int subtaskId) { + assign(subtaskId); + } + + @Override + public synchronized int currentUnassignedSplitSize() { + return pending.size(); + } + + @Override + public synchronized State snapshotState(long checkpointId) { + return new State(consumerSlots, new ArrayList<>(pending.values())); + } + + @Override + public void notifyCheckpointComplete(long checkpointId) {} + + @Override + public void close() {} + + public static final class Split implements SourceSplit { + private static final long serialVersionUID = 1L; + private final int id; + + public Split(int id) { + this.id = id; + } + + @Override + public String splitId() { + return "consumer-" + id; + } + } + + public static final class State implements Serializable { + private static final long serialVersionUID = 1L; + private final int consumerSlots; + private final List<Split> pending; + + private State(int consumerSlots, List<Split> pending) { + this.consumerSlots = consumerSlots; + this.pending = pending; + } + } +} diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceFactory.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceFactory.java new file mode 100644 index 0000000000..41b8476238 --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceFactory.java @@ -0,0 +1,131 @@ +/* + * 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.source; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.source.SeaTunnelSource; +import org.apache.seatunnel.api.source.SourceSplit; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.connector.TableSource; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.TableSourceFactory; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSourceOptions.MessageFormat; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; +import org.apache.seatunnel.format.text.TextDeserializationSchema; + +import com.google.auto.service.AutoService; + +import java.io.Serializable; +import java.net.URI; +import java.net.URISyntaxException; + +import static org.apache.seatunnel.api.configuration.util.Conditions.greaterThan; +import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank; +import static org.apache.seatunnel.api.options.ConnectorCommonOptions.SCHEMA; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PASSWORD; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.QUEUE_NAME; +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.ActivemqSourceOptions.FIELD_DELIMITER; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSourceOptions.FORMAT; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSourceOptions.MAX_IN_FLIGHT_MESSAGES; + +/** Factory for the streaming, checkpoint-aware ActiveMQ Classic queue source. */ +@AutoService(Factory.class) +public class ActivemqSourceFactory implements TableSourceFactory { + + @Override + public String factoryIdentifier() { + return "ActiveMQ"; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(URI, notBlank(URI)) + .required(QUEUE_NAME, notBlank(QUEUE_NAME)) + .required(SCHEMA) + .bundled(USERNAME, PASSWORD) + .optional(FORMAT) + .optional(FIELD_DELIMITER, notBlank(FIELD_DELIMITER)) + .optional(MAX_IN_FLIGHT_MESSAGES, greaterThan(MAX_IN_FLIGHT_MESSAGES, 0)) + .build(); + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + ReadonlyConfig config = context.getOptions(); + validate(config); + CatalogTable table = CatalogTableUtil.buildWithConfig(config); + DeserializationSchema<SeaTunnelRow> deserializer = + config.get(FORMAT) == MessageFormat.JSON + ? new JsonDeserializationSchema(table, false, false) + : TextDeserializationSchema.builder() + .seaTunnelRowType(table.getSeaTunnelRowType()) + .delimiter(config.get(FIELD_DELIMITER)) + .build(); + return () -> + (SeaTunnelSource<T, SplitT, StateT>) + new ActivemqSource(config, table, deserializer); + } + + /** Validate before opening connections, including callers that construct a source directly. */ + static void validate(ReadonlyConfig config) { + ConfigValidator.of(config).validate(new ActivemqSourceFactory().optionRule()); + try { + URI endpoint = new URI(config.get(URI)); + if (!("tcp".equals(endpoint.getScheme()) || "ssl".equals(endpoint.getScheme())) + || endpoint.getHost() == null + || endpoint.getPort() <= 0 + || endpoint.getPort() > 65535 + || endpoint.getUserInfo() != null + || endpoint.getQuery() != null + || endpoint.getFragment() != null + || (endpoint.getPath() != null && !endpoint.getPath().isEmpty())) { + throw new IllegalArgumentException( + "ActiveMQ source uri must be tcp://host:port or ssl://host:port; " + + "use username/password options for credentials. " + + "Composite transports and URI options are not supported."); + } + } catch (URISyntaxException e) { + // Do not include the supplied URI: it may contain credentials. + throw new IllegalArgumentException("Invalid ActiveMQ source uri"); + } + String queue = config.get(QUEUE_NAME); + if (queue.contains(",") + || queue.contains("?") + || queue.contains(">") + || queue.contains("*") + || queue.contains("://")) { + throw new IllegalArgumentException( + "ActiveMQ source queue_name must name one literal queue without destination options"); + } + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return ActivemqSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceReader.java b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceReader.java new file mode 100644 index 0000000000..081b464ca8 --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceReader.java @@ -0,0 +1,366 @@ +/* + * 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.source; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceEnumerator.Split; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; + +import org.apache.activemq.ActiveMQConnection; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.ActiveMQSession; +import org.apache.activemq.transport.InactivityIOException; +import org.apache.activemq.transport.TransportDisposedIOException; + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.Session; +import javax.jms.TextMessage; + +import java.io.IOException; +import java.net.Socket; +import java.net.SocketException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Deque; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicReference; + +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.PASSWORD; +import static org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions.QUEUE_NAME; +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.ActivemqSourceOptions.MAX_IN_FLIGHT_MESSAGES; + +/** + * Synchronous queue reader with checkpoint-scoped individual acknowledgements. JMS calls and local + * state are serialized; record emission also holds the engine checkpoint lock. Only logical + * consumer slots are serialized: a new connection receives unacknowledged messages again from the + * broker. + */ +public class ActivemqSourceReader implements SourceReader<SeaTunnelRow, Split> { + private final ReadonlyConfig config; + private final DeserializationSchema<SeaTunnelRow> deserializer; + private final ConnectionFactory connectionFactory; + private final Map<String, Split> splits = new LinkedHashMap<>(); + private final Deque<PendingMessage> pending = new ArrayDeque<>(); + // Sequence watermarks avoid copying all pending message handles for every checkpoint. + private final NavigableMap<Long, Long> checkpoints = new TreeMap<>(); + private final AtomicReference<Exception> failure = new AtomicReference<>(); + private final int maxInFlight; + private long sequence; + private volatile boolean closed; + private volatile Connection connection; + private Session session; + private MessageConsumer consumer; + + public ActivemqSourceReader( + ReadonlyConfig config, DeserializationSchema<SeaTunnelRow> deserializer) { + this(config, deserializer, createConnectionFactory(config)); + } + + ActivemqSourceReader( + ReadonlyConfig config, + DeserializationSchema<SeaTunnelRow> deserializer, + ConnectionFactory connectionFactory) { + this.config = config; + this.deserializer = deserializer; + this.connectionFactory = connectionFactory; + this.maxInFlight = config.get(MAX_IN_FLIGHT_MESSAGES); + } + + private static ActiveMQConnectionFactory createConnectionFactory(ReadonlyConfig config) { + ActivemqSourceFactory.validate(config); + ActiveMQConnectionFactory factory = + new ActiveMQConnectionFactory(config.get(URI) + "?connectionTimeout=10000"); + factory.getPrefetchPolicy().setQueuePrefetch(config.get(MAX_IN_FLIGHT_MESSAGES)); + factory.setOptimizeAcknowledge(false); + // The checkpoint is already durable. One-way ACKs avoid an untimed broker request per + // message; a lost ACK can only cause redelivery, not loss of checkpointed data. + factory.setSendAcksAsync(true); + factory.setWatchTopicAdvisories(false); + factory.setTrustAllPackages(false); + factory.setTrustedPackages(Collections.emptyList()); + factory.setCloseTimeout(5000); + factory.setConnectResponseTimeout(10000); + return factory; + } + + /** Connections are opened lazily once this reader receives a consumer slot. */ + @Override + public void open() {} + + private void openConsumer() throws JMSException { + if (consumer != null) { + return; + } + try { + connection = + config.getOptional(USERNAME).isPresent() + ? connectionFactory.createConnection( + config.get(USERNAME), config.get(PASSWORD)) + : connectionFactory.createConnection(); + // close() may have observed null while createConnection was in progress. + if (closed) { + throw new JMSException("ActiveMQ source was closed during connection creation"); + } + connection.setExceptionListener(error -> failure.compareAndSet(null, error)); + // CLIENT_ACKNOWLEDGE is cumulative and could acknowledge post-checkpoint records. + session = connection.createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE); + consumer = session.createConsumer(session.createQueue(config.get(QUEUE_NAME))); + connection.start(); + } catch (JMSException e) { + failure.compareAndSet(null, e); + try { + close(); + } catch (IOException closeError) { + e.addSuppressed(closeError); + } + throw e; + } + } + + /** Emit only from pollNext, never from an asynchronous JMS listener. */ + @Override + public void pollNext(Collector<SeaTunnelRow> output) throws Exception { + synchronized (this) { + checkFailure(); + if (closed) { + return; + } + if (splits.isEmpty() || pending.size() >= maxInFlight) { + wait(100); + return; + } + } + + for (int i = 0; i < 32; i++) { + Message message; + synchronized (this) { + checkFailure(); + if (closed || pending.size() >= maxInFlight) { + return; + } + openConsumer(); + message = i == 0 ? consumer.receive(200) : consumer.receiveNoWait(); + } + if (message == null) { + checkFailure(); + return; + } + synchronized (output.getCheckpointLock()) { + synchronized (this) { + checkFailure(); + if (closed) { + return; + } + try { + if (!(message instanceof TextMessage)) { + throw new IOException( + "ActiveMQ source accepts only JMS TextMessage payloads"); + } + String text = ((TextMessage) message).getText(); + if (text == null) { + throw new IOException( + "ActiveMQ source received a null TextMessage payload"); + } + // The shared text deserializer emits no row for an empty payload. Never + // acknowledge a queue message that would be silently dropped. + if (text.isEmpty()) { + throw new IOException( + "ActiveMQ source received an empty TextMessage payload"); + } + try { + byte[] payload = text.getBytes(StandardCharsets.UTF_8); + if (deserializer instanceof JsonDeserializationSchema) { + ((JsonDeserializationSchema) deserializer).collect(payload, output); + } else { + deserializer.deserialize(payload, output); + } + } catch (Exception error) { + // Shared format errors may embed the full queue payload. Neither retain + // their causes nor expose their messages through engine task logs. + throw new IOException( + "ActiveMQ message deserialization or emission failed (" + + error.getClass().getSimpleName() + + "); check the configured schema and format"); + } + pending.addLast(new PendingMessage(++sequence, message)); + } catch (Exception e) { + failure.compareAndSet(null, e); + // No acknowledgement: closing the failed reader makes the message eligible + // for redelivery, including partial multi-row deserialization/emission. + throw e; + } + } + } + } + } + + @Override + public synchronized void addSplits(List<Split> assigned) { + assigned.forEach(split -> splits.putIfAbsent(split.splitId(), split)); + notifyAll(); + } + + @Override + public void handleNoMoreSplits() { + // An empty queue is not end-of-input for this unbounded source. + } + + /** Snapshot emitted records only; received but not emitted records remain broker-owned. */ + @Override + public synchronized List<Split> snapshotState(long checkpointId) throws Exception { + checkFailure(); + checkpoints.put(checkpointId, sequence); + return new ArrayList<>(splits.values()); + } + + /** Acknowledge only records included in this completed checkpoint, including earlier aborts. */ + @Override + public synchronized void notifyCheckpointComplete(long checkpointId) throws Exception { + if (closed) { + return; + } + checkFailure(); + Long watermark = checkpoints.get(checkpointId); + if (watermark == null) { + return; + } + try { + while (!pending.isEmpty() && pending.peekFirst().sequence <= watermark) { + pending.peekFirst().message.acknowledge(); + pending.removeFirst(); + } + } catch (JMSException e) { + failure.compareAndSet(null, e); + throw new IOException( + "ActiveMQ acknowledgement failed for checkpoint " + checkpointId, e); + } + checkpoints.headMap(checkpointId, true).clear(); + notifyAll(); + } + + @Override + public synchronized void notifyCheckpointAborted(long checkpointId) { + checkpoints.remove(checkpointId); + } + + private void checkFailure() throws IOException { + Exception error = failure.get(); + if (error != null) { + throw new IOException( + "ActiveMQ source reader failed; unacknowledged messages will be redelivered", + error); + } + } + + /** Close every JMS resource without acknowledging buffered or emitted records. */ + @Override + public void close() throws IOException { + closed = true; + // Cancellation must not wait for the JMS lock: the broker may be withholding a response + // while its transport remains alive. Closing the TCP/SSL socket unblocks pending I/O. + Connection activeConnection = connection; + IOException socketError = null; + boolean interruptedTransport = false; + if (activeConnection instanceof ActiveMQConnection) { + Socket socket = + ((ActiveMQConnection) activeConnection).getTransport().narrow(Socket.class); + if (socket != null) { + try { + socket.close(); + interruptedTransport = true; + } catch (IOException e) { + socketError = e; + } + } + } + synchronized (this) { + try { + closeResources(interruptedTransport); + } catch (IOException e) { + if (socketError != null) { + e.addSuppressed(socketError); + } + throw e; + } + } + if (socketError != null) { + throw socketError; + } + } + + private void closeResources(boolean interruptedTransport) throws IOException { + try { + if (connection != null) { + // Connection.close cascades to sessions/consumers and disposes local state even + // if its final broker removal command fails after cancellation closed the socket. + connection.close(); + } + } catch (JMSException e) { + if (!interruptedTransport + || !(connection instanceof ActiveMQConnection) + || !((ActiveMQConnection) connection).isClosed() + || !isTransportCloseError(e)) { + throw new IOException("Could not close ActiveMQ connection", e); + } + } finally { + consumer = null; + session = null; + connection = null; + pending.clear(); + checkpoints.clear(); + notifyAll(); + } + } + + private static boolean isTransportCloseError(JMSException error) { + for (Throwable cause = error.getCause(); cause != null; cause = cause.getCause()) { + if (cause instanceof SocketException + || cause instanceof TransportDisposedIOException + || cause instanceof InactivityIOException) { + return true; + } + } + return false; + } + + private static final class PendingMessage { + private final long sequence; + private final Message message; + + private PendingMessage(long sequence, Message message) { + this.sequence = sequence; + this.message = message; + } + } +} 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 56defbb500..2141330010 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 @@ -17,16 +17,23 @@ package org.apache.seatunnel.connectors.seatunnel.activemq; +import org.apache.seatunnel.api.common.JobContext; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.ConfigValidator; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.source.Boundedness; +import org.apache.seatunnel.api.source.SeaTunnelSource; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; +import org.apache.seatunnel.common.constants.JobMode; import org.apache.seatunnel.connectors.seatunnel.activemq.config.ActivemqSinkOptions; import org.apache.seatunnel.connectors.seatunnel.activemq.sink.ActivemqSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -92,6 +99,96 @@ class ActivemqFactoryTest { ConfigValidator.of(readonlyConfig).validate(optionRule); } + @Test + void testSourceOptionsAndFactory() { + ActivemqSourceFactory factory = new ActivemqSourceFactory(); + Map<String, Object> config = sourceConfig(); + ReadonlyConfig options = ReadonlyConfig.fromMap(config); + ConfigValidator.validateUnknownKeys(options, factory.optionRule(), "ActiveMQSource"); + SeaTunnelSource<?, ?, ?> source = createSource(config); + Assertions.assertEquals("ActiveMQ", source.getPluginName()); + Assertions.assertEquals(Boundedness.UNBOUNDED, source.getBoundedness()); + Assertions.assertEquals(1, source.getProducedCatalogTables().size()); + } + + @Test + void testSourceRejectsInvalidOptions() { + for (String key : new String[] {"uri", "queue_name", "field_delimiter"}) { + Map<String, Object> config = sourceConfig(); + config.put(key, " "); + Assertions.assertThrows(OptionValidationException.class, () -> createSource(config)); + } + for (int value : new int[] {0, -1}) { + Map<String, Object> config = sourceConfig(); + config.put("max_in_flight_messages", value); + Assertions.assertThrows(OptionValidationException.class, () -> createSource(config)); + } + for (String key : new String[] {"username", "password"}) { + Map<String, Object> config = sourceConfig(); + config.put(key, "test"); + Assertions.assertThrows(OptionValidationException.class, () -> createSource(config)); + } + } + + @Test + void testSourceRejectsCompositeDestinationsAndTransports() { + for (String uri : + new String[] { + "failover:(tcp://localhost:61616)", + "vm://broker", + "http://localhost:8161";, + "tcp://user:secret@localhost:61616", + "tcp://localhost:61616?jms.prefetchPolicy.all=0", + "tcp://localhost", + "tcp://localhost:65536", + "tcp://localhost:61616/path" + }) { + Map<String, Object> config = sourceConfig(); + config.put("uri", uri); + IllegalArgumentException error = + Assertions.assertThrows( + IllegalArgumentException.class, () -> createSource(config)); + Assertions.assertFalse(error.getMessage().contains("secret")); + } + for (String queue : + new String[] {"a,b", "a?consumer.prefetchSize=0", "a.>", "a.*", "topic://a"}) { + Map<String, Object> config = sourceConfig(); + config.put("queue_name", queue); + Assertions.assertThrows(IllegalArgumentException.class, () -> createSource(config)); + } + } + + @Test + void testSourceRequiresStreamingCheckpoints() { + SeaTunnelSource<?, ?, ?> source = createSource(sourceConfig()); + JobContext context = new JobContext(); + context.setJobMode(JobMode.BATCH); + context.setEnableCheckpoint(true); + source.setJobContext(context); + Assertions.assertThrows(IllegalArgumentException.class, source::getBoundedness); + context.setJobMode(JobMode.STREAMING); + context.setEnableCheckpoint(false); + Assertions.assertThrows(IllegalArgumentException.class, source::getBoundedness); + context.setEnableCheckpoint(true); + Assertions.assertEquals(Boundedness.UNBOUNDED, source.getBoundedness()); + } + + private SeaTunnelSource<?, ?, ?> createSource(Map<String, Object> config) { + return new ActivemqSourceFactory() + .createSource( + new TableSourceFactoryContext( + ReadonlyConfig.fromMap(config), getClass().getClassLoader())) + .createSource(); + } + + private Map<String, Object> sourceConfig() { + Map<String, Object> config = requiredConfig(); + config.put( + "schema", + Collections.singletonMap("fields", Collections.singletonMap("value", "string"))); + return config; + } + private Map<String, Object> requiredConfig() { Map<String, Object> config = new HashMap<>(); config.put(ActivemqSinkOptions.URI.key(), "tcp://localhost:61616"); diff --git a/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceReaderTest.java b/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceReaderTest.java new file mode 100644 index 0000000000..ca2c5eee80 --- /dev/null +++ b/seatunnel-connectors-v2/connector-activemq/src/test/java/org/apache/seatunnel/connectors/seatunnel/activemq/source/ActivemqSourceReaderTest.java @@ -0,0 +1,553 @@ +/* + * 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.source; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.api.table.type.BasicType; +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.activemq.source.ActivemqSourceEnumerator.Split; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceEnumerator.State; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; + +import org.apache.activemq.ActiveMQConnection; +import org.apache.activemq.ActiveMQSession; +import org.apache.activemq.transport.Transport; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.ObjectMessage; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.TextMessage; + +import java.io.IOException; +import java.io.PrintWriter; +import java.io.StringWriter; +import java.net.Socket; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +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.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ActivemqSourceReaderTest { + private ConnectionFactory factory; + private Connection connection; + private Session session; + private MessageConsumer consumer; + private Collector<SeaTunnelRow> collector; + private DeserializationSchema<SeaTunnelRow> deserializer; + private ActivemqSourceReader reader; + private java.util.Queue<Message> messages; + + @BeforeEach + void setUp() throws Exception { + factory = mock(ConnectionFactory.class); + connection = mock(Connection.class); + session = mock(Session.class); + consumer = mock(MessageConsumer.class); + collector = mock(Collector.class); + deserializer = mock(DeserializationSchema.class); + messages = new ArrayDeque<>(); + when(factory.createConnection()).thenReturn(connection); + when(connection.createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE)) + .thenReturn(session); + when(session.createQueue("events")).thenReturn(mock(Queue.class)); + when(session.createConsumer(any())).thenReturn(consumer); + when(consumer.receive(anyLong())).thenAnswer(call -> messages.poll()); + when(consumer.receiveNoWait()).thenAnswer(call -> messages.poll()); + when(collector.getCheckpointLock()).thenReturn(new Object()); + reader = new ActivemqSourceReader(config(), deserializer, factory); + reader.open(); + reader.addSplits(Collections.singletonList(new Split(0))); + } + + @AfterEach + void tearDown() throws Exception { + reader.close(); + } + + @Test + void acknowledgesOnlyTheCompletedSnapshot() throws Exception { + TextMessage first = enqueue("first"); + reader.pollNext(collector); + reader.snapshotState(1); + TextMessage later = enqueue("later"); + reader.pollNext(collector); + verify(first, never()).acknowledge(); + reader.notifyCheckpointComplete(1); + verify(first).acknowledge(); + verify(later, never()).acknowledge(); + reader.snapshotState(2); + reader.notifyCheckpointComplete(2); + verify(later).acknowledge(); + verify(connection).createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE); + } + + @Test + void abortedSnapshotRemainsCoveredByLaterCompletion() throws Exception { + TextMessage first = enqueue("first"); + reader.pollNext(collector); + reader.snapshotState(1); + reader.notifyCheckpointAborted(1); + reader.notifyCheckpointComplete(1); + verify(first, never()).acknowledge(); + reader.snapshotState(2); + reader.notifyCheckpointComplete(2); + verify(first).acknowledge(); + } + + @Test + void completionIsIdempotentAndHandlesSubsumedCheckpoints() throws Exception { + TextMessage first = enqueue("first"); + reader.pollNext(collector); + reader.snapshotState(1); + reader.snapshotState(2); + reader.notifyCheckpointComplete(99); + verify(first, never()).acknowledge(); + reader.notifyCheckpointComplete(2); + reader.notifyCheckpointComplete(1); + reader.notifyCheckpointComplete(2); + verify(first, times(1)).acknowledge(); + } + + @Test + void backpressureStopsReceivingUntilCheckpointCompletes() throws Exception { + TextMessage first = enqueue("first"); + enqueue("second"); + TextMessage third = enqueue("third"); + reader.pollNext(collector); + reader.pollNext(collector); + assertEquals(1, messages.size()); + reader.snapshotState(1); + reader.notifyCheckpointComplete(1); + verify(first).acknowledge(); + reader.pollNext(collector); + assertEquals(0, messages.size()); + verify(third, never()).acknowledge(); + } + + @Test + void acknowledgementFailureStopsFurtherProcessing() throws Exception { + TextMessage first = enqueue("first"); + TextMessage second = enqueue("second"); + reader.pollNext(collector); + reader.snapshotState(1); + doThrow(new JMSException("disconnected")).when(second).acknowledge(); + assertThrows(IOException.class, () -> reader.notifyCheckpointComplete(1)); + verify(first).acknowledge(); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + assertThrows(IOException.class, () -> reader.snapshotState(2)); + } + + @Test + void malformedPayloadIsNotAcknowledged() throws Exception { + TextMessage message = enqueue("invalid"); + doThrow(new IOException("invalid JSON")) + .when(deserializer) + .deserialize(any(byte[].class), eq(collector)); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + verify(message, never()).acknowledge(); + assertThrows(IOException.class, () -> reader.snapshotState(1)); + } + + @Test + void failedMultiRowEmissionIsNotAcknowledged() throws Exception { + TextMessage message = enqueue("rows"); + doAnswer( + call -> { + Collector<SeaTunnelRow> output = call.getArgument(1); + output.collect(new SeaTunnelRow(new Object[] {"first"})); + throw new IOException("second row failed"); + }) + .when(deserializer) + .deserialize(any(byte[].class), eq(collector)); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + verify(collector).collect(any(SeaTunnelRow.class)); + verify(message, never()).acknowledge(); + } + + @Test + void rejectsObjectMessagesWithoutDeserializingThem() throws Exception { + ObjectMessage message = mock(ObjectMessage.class); + messages.add(message); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + verify(message, never()).getObject(); + verify(message, never()).acknowledge(); + } + + @Test + void rejectsNullTextPayload() throws Exception { + TextMessage message = enqueue(null); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + verify(message, never()).acknowledge(); + } + + @Test + void rejectsEmptyTextPayloadWithoutAcknowledging() throws Exception { + TextMessage message = enqueue(""); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + verify(deserializer, never()).deserialize(any(byte[].class), eq(collector)); + verify(message, never()).acknowledge(); + assertThrows(IOException.class, () -> reader.snapshotState(1)); + } + + @Test + void closeDoesNotAcknowledgeAndClosesAllResources() throws Exception { + TextMessage message = enqueue("pending"); + reader.pollNext(collector); + reader.snapshotState(1); + reader.close(); + reader.notifyCheckpointComplete(1); + reader.close(); + verify(message, never()).acknowledge(); + // JMS Connection.close owns cascading cleanup without separate broker removal requests. + verify(consumer, never()).close(); + verify(session, never()).close(); + verify(connection, times(1)).close(); + } + + @Test + void connectionCleanupFailureIsReported() throws Exception { + reader.pollNext(collector); + doThrow(new JMSException("connection close failed")).when(connection).close(); + assertThrows(IOException.class, reader::close); + verify(connection).close(); + } + + @Test + void failedStartupClosesPartialResources() throws Exception { + doThrow(new JMSException("start failed")).when(connection).start(); + assertThrows(JMSException.class, () -> reader.pollNext(collector)); + verify(connection).close(); + } + + @Test + void asynchronousConnectionFailureIsPropagated() throws Exception { + reader.pollNext(collector); + ArgumentCaptor<ExceptionListener> listener = + ArgumentCaptor.forClass(ExceptionListener.class); + verify(connection).setExceptionListener(listener.capture()); + listener.getValue().onException(new JMSException("broker stopped")); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + } + + @Test + void duplicateRestoredSlotsDoNotCreateExtraConsumers() throws Exception { + reader.addSplits(Arrays.asList(new Split(0), new Split(1))); + reader.pollNext(collector); + assertEquals(2, reader.snapshotState(1).size()); + verify(session, times(1)).createConsumer(any()); + } + + @Test + void unassignedReaderDoesNotConnect() throws Exception { + reader.close(); + reader = new ActivemqSourceReader(config(), deserializer, factory); + reader.open(); + reader.handleNoMoreSplits(); + reader.pollNext(collector); + assertEquals(0, reader.snapshotState(1).size()); + verify(factory, never()).createConnection(); + } + + @Test + void enumeratorRestoresOnlyPendingSlotsAndReassignsReturnedSlots() throws Exception { + SourceSplitEnumerator.Context<Split> context = mock(SourceSplitEnumerator.Context.class); + when(context.currentParallelism()).thenReturn(2); + when(context.registeredReaders()).thenReturn(Collections.singleton(0)); + ActivemqSourceEnumerator enumerator = new ActivemqSourceEnumerator(context); + enumerator.run(); + assertEquals(1, enumerator.currentUnassignedSplitSize()); + State state = enumerator.snapshotState(1); + SourceSplitEnumerator.Context<Split> restoredContext = + mock(SourceSplitEnumerator.Context.class); + when(restoredContext.currentParallelism()).thenReturn(2); + when(restoredContext.registeredReaders()).thenReturn(new HashSet<>(Arrays.asList(0, 1))); + ActivemqSourceEnumerator restored = new ActivemqSourceEnumerator(restoredContext, state); + restored.run(); + verify(restoredContext, never()).assignSplit(eq(0), any(List.class)); + verify(restoredContext).assignSplit(eq(1), any(List.class)); + assertEquals(0, restored.currentUnassignedSplitSize()); + restored.addSplitsBack(Collections.singletonList(new Split(0)), 0); + verify(restoredContext).assignSplit(eq(0), any(List.class)); + // State snapshots must not alias the live pending collection. + ActivemqSourceEnumerator anotherRestore = + new ActivemqSourceEnumerator(restoredContext, state); + assertEquals(1, anotherRestore.currentUnassignedSplitSize()); + } + + @Test + void failedAssignmentDoesNotLosePendingSlots() { + SourceSplitEnumerator.Context<Split> context = mock(SourceSplitEnumerator.Context.class); + when(context.currentParallelism()).thenReturn(1); + when(context.registeredReaders()).thenReturn(Collections.singleton(0)); + doThrow(new IllegalStateException("reader unavailable")) + .when(context) + .assignSplit(eq(0), any(List.class)); + ActivemqSourceEnumerator enumerator = new ActivemqSourceEnumerator(context); + assertThrows(IllegalStateException.class, enumerator::run); + assertEquals(1, enumerator.currentUnassignedSplitSize()); + } + + @Test + void restoreScaleUpAddsOnlyNewSlots() throws Exception { + SourceSplitEnumerator.Context<Split> context = mock(SourceSplitEnumerator.Context.class); + when(context.currentParallelism()).thenReturn(1); + when(context.registeredReaders()).thenReturn(Collections.singleton(0)); + ActivemqSourceEnumerator enumerator = new ActivemqSourceEnumerator(context); + enumerator.run(); + State saved = enumerator.snapshotState(1); + SourceSplitEnumerator.Context<Split> restoredContext = + mock(SourceSplitEnumerator.Context.class); + when(restoredContext.currentParallelism()).thenReturn(4); + when(restoredContext.registeredReaders()) + .thenReturn(new HashSet<>(Arrays.asList(0, 1, 2, 3))); + ActivemqSourceEnumerator restored = new ActivemqSourceEnumerator(restoredContext, saved); + restored.addSplitsBack(Collections.singletonList(new Split(0)), 0); + restored.run(); + restored.run(); + for (int i = 0; i < 4; i++) { + verify(restoredContext, times(1)).assignSplit(eq(i), any(List.class)); + } + assertEquals(0, restored.currentUnassignedSplitSize()); + } + + @Test + void credentialsArePassedSeparatelyToConnectionFactory() throws Exception { + reader.close(); + Map<String, Object> options = new HashMap<>(); + options.put("queue_name", "events"); + options.put("username", "reader"); + options.put("password", "test-password"); + when(factory.createConnection("reader", "test-password")).thenReturn(connection); + reader = new ActivemqSourceReader(ReadonlyConfig.fromMap(options), deserializer, factory); + reader.addSplits(Collections.singletonList(new Split(0))); + reader.pollNext(collector); + verify(factory).createConnection("reader", "test-password"); + verify(factory, never()).createConnection(); + } + + @Test + void cancellationInterruptsTransportBeforeWaitingForReaderLock() throws Exception { + ActiveMQConnection activeConnection = mock(ActiveMQConnection.class); + Transport transport = mock(Transport.class); + Socket socket = mock(Socket.class); + when(factory.createConnection()).thenReturn(activeConnection); + when(activeConnection.createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE)) + .thenReturn(session); + when(activeConnection.getTransport()).thenReturn(transport); + when(transport.narrow(Socket.class)).thenReturn(socket); + CountDownLatch receiving = new CountDownLatch(1); + CountDownLatch socketClosed = new CountDownLatch(1); + when(consumer.receive(anyLong())) + .thenAnswer( + call -> { + receiving.countDown(); + assertTrue(socketClosed.await(10, TimeUnit.SECONDS)); + return null; + }); + doAnswer( + call -> { + socketClosed.countDown(); + return null; + }) + .when(socket) + .close(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<?> polling = + executor.submit( + () -> { + reader.pollNext(collector); + return null; + }); + assertTrue(receiving.await(5, TimeUnit.SECONDS)); + Future<?> closing = + executor.submit( + () -> { + reader.close(); + return null; + }); + closing.get(5, TimeUnit.SECONDS); + polling.get(5, TimeUnit.SECONDS); + verify(socket).close(); + verify(activeConnection).close(); + } finally { + socketClosed.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void cancellationDuringConnectionCreationClosesNewConnectionWithoutOpeningSession() + throws Exception { + when(factory.createConnection()) + .thenAnswer( + call -> { + // Reproduce close observing null before the newly created connection is + // published. + reader.close(); + return connection; + }); + assertThrows(JMSException.class, () -> reader.pollNext(collector)); + verify(connection, never()).createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE); + verify(connection).close(); + } + + @Test + void realJsonParserFailureDoesNotExposePayloadOrAcknowledgeMessage() throws Exception { + reader.close(); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"value"}, new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE}); + reader = + new ActivemqSourceReader( + config(), new JsonDeserializationSchema(false, false, rowType), factory); + reader.addSplits(Collections.singletonList(new Split(0))); + String marker = "SYNTHETIC_SECRET_7f83"; + TextMessage message = enqueue("{\"value\":\"" + marker + "\",BROKEN}"); + IOException initial = assertThrows(IOException.class, () -> reader.pollNext(collector)); + IOException subsequent = assertThrows(IOException.class, () -> reader.snapshotState(1)); + for (IOException error : Arrays.asList(initial, subsequent)) { + StringWriter trace = new StringWriter(); + error.printStackTrace(new PrintWriter(trace)); + assertFalse(trace.toString().contains(marker)); + assertTrue(trace.toString().contains("deserialization or emission failed")); + } + verify(message, never()).acknowledge(); + } + + @Test + void jsonArrayAcknowledgesOnlyAfterEveryRowIsEmittedAndCheckpointCompletes() throws Exception { + reader.close(); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"value"}, new SeaTunnelDataType<?>[] {BasicType.LONG_TYPE}); + reader = + new ActivemqSourceReader( + config(), new JsonDeserializationSchema(false, false, rowType), factory); + reader.addSplits(Collections.singletonList(new Split(0))); + TextMessage valid = enqueue("[{\"value\":1},{\"value\":2}]"); + reader.pollNext(collector); + verify(collector, times(2)).collect(any(SeaTunnelRow.class)); + verify(valid, never()).acknowledge(); + reader.snapshotState(1); + reader.notifyCheckpointComplete(1); + verify(valid).acknowledge(); + TextMessage invalid = enqueue("[{\"value\":3},{\"value\":\"not-a-number\"}]"); + assertThrows(IOException.class, () -> reader.pollNext(collector)); + verify(collector, times(3)).collect(any(SeaTunnelRow.class)); + verify(invalid, never()).acknowledge(); + assertThrows(IOException.class, () -> reader.snapshotState(2)); + } + + @Test + void interruptedTransportErrorIsIgnoredOnlyAfterConnectionHasClosed() throws Exception { + ActiveMQConnection active = useActiveConnection(); + reader.pollNext(collector); + JMSException error = new JMSException("socket closed"); + error.initCause(new java.net.SocketException("socket closed")); + doThrow(error).when(active).close(); + when(active.isClosed()).thenReturn(true); + reader.close(); + verify(active).close(); + } + + @Test + void unrelatedCleanupErrorAfterSocketInterruptionIsStillReported() throws Exception { + ActiveMQConnection active = useActiveConnection(); + reader.pollNext(collector); + doThrow(new JMSException("scheduler shutdown failed")).when(active).close(); + when(active.isClosed()).thenReturn(true); + assertThrows(IOException.class, reader::close); + } + + @Test + void transportErrorBeforeConnectionCleanupCompletesIsStillReported() throws Exception { + ActiveMQConnection active = useActiveConnection(); + reader.pollNext(collector); + JMSException error = new JMSException("socket closed"); + error.initCause(new java.net.SocketException("socket closed")); + doThrow(error).when(active).close(); + when(active.isClosed()).thenReturn(false); + assertThrows(IOException.class, reader::close); + } + + private ActiveMQConnection useActiveConnection() throws Exception { + ActiveMQConnection active = mock(ActiveMQConnection.class); + Transport transport = mock(Transport.class); + when(active.getTransport()).thenReturn(transport); + when(transport.narrow(Socket.class)).thenReturn(mock(Socket.class)); + when(factory.createConnection()).thenReturn(active); + when(active.createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE)) + .thenReturn(session); + return active; + } + + private TextMessage enqueue(String text) throws JMSException { + TextMessage message = mock(TextMessage.class); + when(message.getText()).thenReturn(text); + messages.add(message); + return message; + } + + private ReadonlyConfig config() { + Map<String, Object> config = new HashMap<>(); + config.put("queue_name", "events"); + config.put("max_in_flight_messages", 2); + return ReadonlyConfig.fromMap(config); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/activemq/ActivemqIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/activemq/ActivemqIT.java index 56b1faeab8..ced6677b74 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/activemq/ActivemqIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/activemq/ActivemqIT.java @@ -17,8 +17,20 @@ package org.apache.seatunnel.e2e.connector.activemq; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSource; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceEnumerator.Split; +import org.apache.seatunnel.connectors.seatunnel.activemq.source.ActivemqSourceFactory; +import org.apache.seatunnel.e2e.common.TestResource; import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.EngineType; import org.apache.seatunnel.e2e.common.container.TestContainer; +import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer; +import org.apache.seatunnel.e2e.common.util.JobIdGenerator; import org.apache.activemq.ActiveMQConnectionFactory; @@ -32,9 +44,12 @@ import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy; import org.testcontainers.utility.DockerImageName; +import lombok.extern.slf4j.Slf4j; + import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.JMSException; +import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Queue; @@ -43,14 +58,26 @@ import javax.jms.TextMessage; import java.io.IOException; import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; -public class ActivemqIT extends TestSuiteBase { +@Slf4j +public class ActivemqIT extends TestSuiteBase implements TestResource { private static final String ACTIVEMQ_CONTAINER_HOST = "activemq-host"; public GenericContainer<?> activeMQContainer = - new GenericContainer<>(DockerImageName.parse("rmohr/activemq")) + new GenericContainer<>(DockerImageName.parse("apache/activemq-classic:5.18.7")) .withExposedPorts(61616) .withNetworkAliases(ACTIVEMQ_CONTAINER_HOST) .withNetwork(NETWORK); @@ -61,12 +88,13 @@ public class ActivemqIT extends TestSuiteBase { private MessageConsumer consumer; @BeforeAll - public void setup() throws JMSException, InterruptedException { + @Override + public void startUp() throws JMSException, InterruptedException { activeMQContainer .withNetwork(NETWORK) .waitingFor(new HostPortWaitStrategy().withStartupTimeout(Duration.ofMinutes(2))); activeMQContainer.start(); - String brokerUrl = "tcp://127.0.0.1:" + activeMQContainer.getMappedPort(61616); + String brokerUrl = brokerUrl(); ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl); connection = connectionFactory.createConnection(); connection.start(); @@ -83,11 +111,16 @@ public class ActivemqIT extends TestSuiteBase { } @AfterAll + @Override public void tearDown() throws JMSException { - // Cleaning up resources - if (producer != null) producer.close(); - if (session != null) session.close(); - if (connection != null) connection.close(); + try { + if (connection != null) { + // Closing a connection closes its sessions, consumers and producers as well. + connection.close(); + } + } finally { + activeMQContainer.close(); + } } @Test @@ -108,11 +141,244 @@ public class ActivemqIT extends TestSuiteBase { public void testSinkApacheActivemq(TestContainer container) throws IOException, InterruptedException, JMSException { Container.ExecResult execResult = container.executeJob("/fake_source_to_sink.conf"); - TextMessage textMessage = (TextMessage) consumer.receive(); + TextMessage textMessage = (TextMessage) consumer.receive(10000); + Assertions.assertNotNull(textMessage, "Sink did not publish a message"); Assertions.assertTrue(textMessage.getText().contains("map")); Assertions.assertTrue(textMessage.getText().contains("c_boolean")); Assertions.assertTrue(textMessage.getText().contains("c_tinyint")); Assertions.assertTrue(textMessage.getText().contains("c_timestamp")); Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); } + + @Test + public void testSourceCheckpointDoesNotAcknowledgeLaterMessages() throws Exception { + String queue = newQueue(); + publish(queue, "{\"value\":\"before\"}"); + RowCollector rows = new RowCollector(); + try (SourceReader<SeaTunnelRow, Split> reader = createReader(queue, "JSON")) { + awaitRows(reader, rows, 1); + reader.snapshotState(1); + publish(queue, "{\"value\":\"after\"}"); + awaitRows(reader, rows, 2); + reader.notifyCheckpointComplete(1); + } + assertEquals("{\"value\":\"after\"}", receive(queue)); + assertNull(receive(queue), "The completed-checkpoint message was not acknowledged"); + } + + @Test + public void testSourceAbortCloseAndRestoreRedelivers() throws Exception { + String queue = newQueue(); + publish(queue, "{\"value\":\"replay\"}"); + List<Split> state; + try (SourceReader<SeaTunnelRow, Split> reader = createReader(queue, "JSON")) { + awaitRows(reader, new RowCollector(), 1); + state = reader.snapshotState(1); + reader.notifyCheckpointAborted(1); + } + try (SourceReader<SeaTunnelRow, Split> restored = createReader(queue, "JSON")) { + restored.addSplits(state); + RowCollector rows = new RowCollector(); + awaitRows(restored, rows, 1); + assertEquals("replay", rows.rows.get(0).getField(0)); + restored.snapshotState(2); + restored.notifyCheckpointComplete(2); + } + assertNull(receive(queue)); + } + + @Test + public void testSourceParallelConsumersAndTextDeserialization() throws Exception { + String queue = newQueue(); + try (SourceReader<SeaTunnelRow, Split> first = createReader(queue, "TEXT", 1); + SourceReader<SeaTunnelRow, Split> second = createReader(queue, "TEXT", 1)) { + RowCollector firstRows = new RowCollector(); + RowCollector secondRows = new RowCollector(); + // Register both consumers before publishing; otherwise the first consumer can + // legitimately prefetch the entire queue before its competitor connects. + first.pollNext(firstRows); + second.pollNext(secondRows); + publish(queue, "first"); + publish(queue, "second"); + awaitRows(first, firstRows, 1); + awaitRows(second, secondRows, 1); + Assertions.assertNotEquals( + firstRows.rows.get(0).getField(0), secondRows.rows.get(0).getField(0)); + first.snapshotState(1); + second.snapshotState(1); + first.notifyCheckpointComplete(1); + second.notifyCheckpointComplete(1); + } + assertNull(receive(queue)); + } + + @Test + public void testSourceMalformedMessageRemainsOnQueue() throws Exception { + String queue = newQueue(); + publish(queue, "not-json"); + try (SourceReader<SeaTunnelRow, Split> reader = createReader(queue, "JSON")) { + assertThrows( + Exception.class, + () -> { + RowCollector rows = new RowCollector(); + awaitRows(reader, rows, 1); + }); + } + assertEquals("not-json", receive(queue)); + } + + @Test + public void testSourceEmptyTextMessageRemainsOnQueue() throws Exception { + String queue = newQueue(); + publish(queue, ""); + try (SourceReader<SeaTunnelRow, Split> reader = createReader(queue, "TEXT")) { + assertThrows(IOException.class, () -> reader.pollNext(new RowCollector())); + } + assertEquals("", receive(queue)); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.FLINK, EngineType.SPARK}, + disabledReason = + "This streaming lifecycle test uses Zeta job status and checkpoint REST APIs") + public void testSourceStreamingJob(TestContainer container) throws Exception { + String jobId = String.valueOf(JobIdGenerator.newJobId()); + CompletableFuture<Container.ExecResult> job = + CompletableFuture.supplyAsync( + () -> { + try { + return container.executeJob( + "/activemq_source_to_console.conf", jobId); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + try { + await().atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> { + assertStreaming(job); + assertEquals("RUNNING", container.getJobStatus(jobId)); + }); + String value = "activemq-source-" + UUID.randomUUID(); + publish("source-events", "{\"value\":\"" + value + "\"}"); + await().atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> { + assertStreaming(job); + Assertions.assertTrue(container.getServerLogs().contains(value)); + }); + // Count from AFTER observing output so the next completed checkpoint covers the row. + long completed = container.getCompletedCheckpointCount(jobId); + await().atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> { + assertStreaming(job); + Assertions.assertTrue( + container.getCompletedCheckpointCount(jobId) > completed); + }); + } finally { + if (!job.isDone()) { + Container.ExecResult cancelled = container.cancelJob(jobId); + assertEquals(0, cancelled.getExitCode(), cancelled.getStderr()); + } + Container.ExecResult result = job.get(120, TimeUnit.SECONDS); + assertEquals(0, result.getExitCode(), result.getStderr()); + } + assertNull(receive("source-events"), "The source did not acknowledge the checkpointed row"); + } + + private void assertStreaming(CompletableFuture<Container.ExecResult> job) throws Exception { + if (job.isDone()) { + Assertions.fail("Streaming job ended before cancellation: " + job.get().getStderr()); + } + } + + private SourceReader<SeaTunnelRow, Split> createReader(String queue, String format) + throws Exception { + return createReader(queue, format, 10); + } + + private SourceReader<SeaTunnelRow, Split> createReader( + String queue, String format, int inFlight) throws Exception { + Map<String, Object> options = new HashMap<>(); + options.put("uri", brokerUrl()); + options.put("queue_name", queue); + options.put("format", format); + options.put("max_in_flight_messages", inFlight); + options.put( + "schema", + Collections.singletonMap("fields", Collections.singletonMap("value", "string"))); + ActivemqSource source = + (ActivemqSource) + new ActivemqSourceFactory() + .<SeaTunnelRow, Split, + org.apache.seatunnel.connectors.seatunnel.activemq.source + .ActivemqSourceEnumerator.State> + createSource( + new TableSourceFactoryContext( + ReadonlyConfig.fromMap(options), + getClass().getClassLoader())) + .createSource(); + SourceReader<SeaTunnelRow, Split> reader = source.createReader(null); + reader.open(); + reader.addSplits(Collections.singletonList(new Split(0))); + return reader; + } + + private void awaitRows(SourceReader<SeaTunnelRow, Split> reader, RowCollector rows, int count) + throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (rows.rows.size() < count && System.nanoTime() < deadline) { + reader.pollNext(rows); + } + assertEquals(count, rows.rows.size()); + } + + private String brokerUrl() { + return "tcp://" + + activeMQContainer.getHost() + + ":" + + activeMQContainer.getMappedPort(61616); + } + + private String newQueue() { + return "source-" + UUID.randomUUID(); + } + + private void publish(String queue, String text) throws JMSException { + MessageProducer sender = session.createProducer(session.createQueue(queue)); + try { + sender.send(session.createTextMessage(text)); + } finally { + sender.close(); + } + } + + private String receive(String queue) throws JMSException { + MessageConsumer receiver = session.createConsumer(session.createQueue(queue)); + try { + Message message = receiver.receive(1000); + return message == null ? null : ((TextMessage) message).getText(); + } finally { + receiver.close(); + } + } + + private static final class RowCollector implements Collector<SeaTunnelRow> { + private final List<SeaTunnelRow> rows = new ArrayList<>(); + private final Object lock = new Object(); + + @Override + public void collect(SeaTunnelRow row) { + rows.add(row); + } + + @Override + public Object getCheckpointLock() { + return lock; + } + } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/resources/activemq_source_to_console.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/resources/activemq_source_to_console.conf new file mode 100644 index 0000000000..c3bbacd9eb --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e/src/test/resources/activemq_source_to_console.conf @@ -0,0 +1,39 @@ +# +# 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. +# + +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 1000 +} + +source { + ActiveMQ { + uri = "tcp://activemq-host:61616" + queue_name = "source-events" + max_in_flight_messages = 10 + schema { + fields { + value = string + } + } + } +} + +sink { + Console {} +}
