goutamadwant opened a new pull request, #12341:
URL: https://github.com/apache/seatunnel/pull/12341
### Purpose of this pull request
Related to #10753. Adds an ActiveMQ Classic queue source to the existing
connector-activemq module. This is a scoped OpenWire implementation, not
completion of the generic JMS/AMQP request.
The source is absent from current dev, and no competing open ActiveMQ source
PR was found. The earlier source PR #8581 was closed without merging.
- Read JSON objects/arrays or delimited text from JMS TextMessages using
existing format implementations.
- Support parallel competing consumers with checkpoint-scoped individual
acknowledgements and at-least-once delivery.
- Bound in-flight messages and preserve unacknowledged messages across
aborted checkpoints and recovery.
- Handle rescaling, cancellation, partial startup/cleanup, connection
failures and partial multi-row emission.
- Prevent payload-bearing parser exceptions from leaking message contents
into task logs.
- Add source registration, English/Chinese documentation, lifecycle tests
and broker-backed E2E coverage.
- Pin the E2E broker image to apache/activemq-classic:5.18.7.
### Does this PR introduce _any_ user-facing change?
Yes. ActiveMQ can now be configured as a streaming source with checkpointing
enabled. Previously this module provided only a sink. Existing sink runtime
behavior, options and dependencies are unchanged.
The source supports one literal queue over TCP or SSL, optional separate
credentials, JSON/TEXT formats and configurable max_in_flight_messages.
Delivery is at least once, not exactly once; broker persistence, expiration and
redelivery policies still apply.
Topics, selectors, composite/failover transports, generic JMS/AMQP and Spark
source execution are outside this slice.
### How was this patch tested?
- 36 connector tests passed on Java 8 and Java 11.
- All 10 ActivemqIT cases passed with Java 8, including broker-backed source
tests, Zeta source checkpoint/acknowledgement/cancellation, and existing sink
tests on Flink 1.18/1.20, Spark 3.3 and Zeta.
- All five source-specific broker/Zeta cases also passed with Java 11.
- Connector installation, starter/reactor compilation and Spotless checks
passed.
Unit tests: `./mvnw -pl seatunnel-connectors-v2/connector-activemq test`
After building the standard E2E prerequisites, run with Docker available:
`RUN_ALL_CONTAINER=false ./mvnw -pl
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-activemq-e2e -DskipIT=false
-Dit.test=ActivemqIT verify`
Validation limits: TLS against a secured broker and source restore under
Flink were not exercised end to end. The Flink/Spark results above cover the
existing sink, not the new source.
### Check list
* [ ] New JAR dependency/license notice: not applicable; existing
dependencies and connector artifact are reused.
* [x] Documentation updated in English and Chinese.
* [ ] incompatible-changes.md: not applicable; additive source with
unchanged sink behavior.
* [x] Connector integration checked:
1. plugin-mapping.properties: source mapping added.
2. seatunnel-dist/pom.xml: existing ActiveMQ registration verified.
3. label-scope-conf.yml: existing ActiveMQ CI scope verified.
4. seatunnel-e2e: existing ActiveMQ IT extended.
5. config/plugin_config: existing connector entry verified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
