SEPURI-SAI-KRISHNA opened a new issue, #13078: URL: https://github.com/apache/seatunnel/issues/13078
### Search before asking - [X] I had searched in the [feature](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Feature%22) and found no similar feature requirement. ### Description ## Overview Design proposal for the **JMS / AMQP (generic)** row of #10753. The tracker lists it as "Partial" with no owner, and notes that the ActiveMQ Classic source from #12341 does not complete the generic request. An earlier status refresh in that thread described the row as "pending scope clarification". The proposal is a new `connector-jms` module, plugin name `Jms`, that reads from and writes to any broker through the standard JMS 1.1 API (`javax.jms`), using the broker's own client library. Section 1 proposes the scope this row needs. Section 4 is the core of the design: standard JMS has no per-message acknowledgement, so the source needs a different checkpoint design from #12341. ## 1. Scope - **In:** any provider with a `javax.jms` `ConnectionFactory`. Examples: ActiveMQ Artemis, ActiveMQ Classic, IBM MQ, TIBCO EMS, Solace. AMQP 1.0 brokers are reached through Apache Qpid JMS, which is a `javax.jms` client. - **Unchanged:** `connector-rabbitmq` keeps AMQP 0-9-1, and `connector-activemq` keeps its ActiveMQ-specific features, such as per-message acknowledgement. This connector neither replaces nor deprecates them. - **API level:** JMS 1.1 only. Providers that support only the `jakarta.jms` namespace are out of scope for v1. ## 2. Delivery in slices 1. Module, connection factory, and the queue source (sections 3 and 4). 2. The sink (section 5). 3. Later, if wanted: durable topic subscriptions, a bounded "drain the queue" mode, `jakarta.jms`. ## 3. Connection factory Two ways to obtain the `ConnectionFactory`. `connection_factory_class` and `jndi_connection_factory_name` are declared with `OptionRule.exclusive`, so exactly one of them must be set. - **Class and properties:** `connection_factory_class` plus `connection_factory_properties` (a map). The class is created with its no-argument constructor, and each property is applied through its setter. The value is converted to the setter's parameter type (`String`, `int`, `long` or `boolean`, or their boxed forms). An unknown property or a failed conversion fails when the reader or writer creates the factory, and the error names the property but not its value. Apache NiFi's `JMSConnectionFactoryProvider` takes the same approach: an implementation class and properties applied through setters. - **JNDI:** `jndi_initial_context_factory`, `jndi_provider_url`, `jndi_connection_factory_name` and `jndi_properties`. - `username` and `password`, when set, are passed to `createConnection(username, password)`; otherwise `createConnection()` is used. - **Client jars:** supplied by the user and not bundled, as with JDBC drivers, because their licences and versions vary by vendor. They go in the connector's isolated dependency directory, `${SEATUNNEL_HOME}/plugins/connector-jms/`, so they load with the connector. The docs will give examples for Artemis, IBM MQ and Qpid JMS. ## 4. Source: checkpoint design **Constraints.** - A queue has no offset to store in state. - `Message.acknowledge()` is cumulative: "a client acknowledges all messages consumed by the session that the message was delivered to". `ActivemqSourceReader` notes this ("CLIENT_ACKNOWLEDGE is cumulative and could acknowledge post-checkpoint records") and uses `ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE`, which is ActiveMQ-specific. - Sessions "are designed for serial use by one thread at a time". The spec allows several threads to use a session as long as they never use it concurrently, with the application providing the synchronization. - In Zeta: - `snapshotState` runs while the barrier holds the checkpoint lock. - `notifyCheckpointComplete` is delivered by `CheckpointFinishedOperation` on another thread, without that lock. - Once a closing barrier (stop with savepoint, job end) has passed, `pollNext` is no longer called, but `notifyCheckpointComplete` for that checkpoint is still delivered before the task closes. - Per the review of #8581, `pollNext` must not keep producing data after it returns, so an asynchronous `MessageListener` is ruled out. **Design.** 1. **One transacted `Session` per checkpoint epoch, on one `Connection` per reader.** A commit then covers exactly one epoch's messages. `Connection` supports concurrent use. 2. **Atomic receive and emit.** Inside `pollNext`, under the checkpoint lock, the reader calls `receiveNoWait()` and emits the row. A message can therefore never be received before a barrier and emitted after it. When nothing is received, `pollNext` returns, and Zeta's idle wait happens outside the lock, so an empty queue never delays a barrier (the concern raised on #12910 about `receive_timeout`). 3. **`snapshotState(N)` only marks the open epoch as sealed for checkpoint N.** It makes no JMS call. The state stores only the logical consumer-slot ids, as in #12341. 4. **Before receiving again, `pollNext` closes the sealed epoch's consumer and opens the next epoch's session and consumer.** The sealed session stays uncommitted. 5. **`notifyCheckpointComplete(N)` commits and closes every sealed session with an id up to N.** It runs on the notifying thread, so it also covers the final checkpoint after `pollNext` has stopped. A per-reader lock serializes every session operation between this thread and the reader thread, so no session is ever used concurrently. If the notification arrives before the reader has moved to the next epoch, the session is still committed and closed, and the reader's later close of that consumer is ignored, as the spec requires for a closed session. Commits run outside the checkpoint lock, and a repeated notification for the same id is a no-op. 6. **An aborted checkpoint commits nothing.** A later completed checkpoint commits the earlier epochs too. 7. **On failure, cancellation or restore, uncommitted sessions are closed or their connection is lost.** The spec says "Closing a transacted session must roll back the transaction in progress", so the broker redelivers those messages. Delivery is at least once. Duplicates are possible after a failure, and order is not guaranteed across a redelivery. 8. **The number of received but uncommitted messages per reader is bounded,** as `max_in_flight_messages` is in #12341. At the limit, the reader stops receiving until commits catch up. **Spike results.** I checked this design before proposing it. - **Setup:** - three brokers in-JVM: ActiveMQ Classic over OpenWire, Artemis over its core protocol, and Artemis over AMQP 1.0 through Qpid JMS; - code compiled against the Geronimo JMS 1.1 API only; - run on JDK 11 and 17. - **Results on all three:** - acknowledging one message also acknowledged the other one received in the same session; - a sealed epoch was committed from a second thread, started while the reader was polling the next epoch's session on the same connection, and when the connection was then dropped, only the uncommitted epoch's message came back; - a session closed without commit returned its message; - committing and closing a session before the reader moved on worked, and the reader's later `consumer.close()` was ignored. - **Negative control:** with one transacted session, a message received after the snapshot was committed by that checkpoint and was gone after a failure. I can share the spike code, or turn it into the first tests of slice 1. **Other rules.** - Parallelism means competing consumers on one queue, assigned as logical slots as in #12341. - `message_selector` is supported, because it is standard JMS. - Rows follow a user-declared `schema`, as in the ActiveMQ source. `TextMessage` and `BytesMessage` payloads go through the existing `json` and `text` formats. Any other message type fails the task and is reported by message id only. - Payloads are never written to logs. - `ObjectMessage` is never deserialized, because that runs Java deserialization on untrusted data. - The source is streaming only. Zeta is the target and is covered by the E2E. Flink will be added to the E2E if the translation layer supports it, and the docs will state only what the E2E covers. The Spark source is out of v1, as in #12341. ## 5. Sink (slice 2) - One transacted producer session per writer, committed in `prepareCommit(checkpointId)`. This is at least once: a failure after the commit and before the checkpoint completes replays rows, so duplicates are possible. Exactly once would need XA, which is out of scope. - Options: `destination`, `message_type` (`TEXT` or `BYTES`), `format`, `delivery_mode`, `priority`, and `time_to_live` (`long`). ## 6. Secret boundary - `password`, the values of `jndi_properties` and the values of `connection_factory_properties` are handed only to the provider (`createConnection`, the `InitialContext` environment, setters). They are never logged and never put in error messages; errors name the key only. - The existing config shade encrypts top-level options, and `username` and `password` are among its default keywords. It does not process values inside a map option, so the docs will recommend keeping credentials in `username` and `password` rather than in the property maps. - Message payloads are not logged. Checkpoint state holds no credentials, only slot ids. ## 7. Dependencies, licence and shading - The only new compile dependency is `org.apache.geronimo.specs:geronimo-jms_1.1_spec:1.1.1`, which is Apache License 2.0. The same artifact and version are already in the build through `activemq-client` 5.15.16 in `connector-activemq`. - `jakarta.jms-api` (EPL 2.0 / GPL 2.0 with Classpath Exception) and `javax.jms-api` (CDDL / GPL) are not used. - No vendor client is bundled. - Vendor clients bring their own `javax.jms` API jar: `geronimo-jms_1.1_spec` for ActiveMQ Classic, `jakarta.jms-api` 2.0.3 for the Artemis 2.19.1 client, the alpha `geronimo-jms_2.0_spec` for Qpid JMS 0.61.0, and `javax.jms-api` 2.0.1 for IBM MQ `allclient`. In the spike, the code compiled against JMS 1.1 worked with all three clients on one class path, whether the Geronimo 1.1 jar or `jakarta.jms-api` 2.0.3 came first. Slice 1 will verify the same inside SeaTunnel's plugin class loader on Zeta, and on Flink if covered. - Test scope only, all Apache License 2.0, and all built for Java 8: `activemq-broker` 5.15.16, `artemis-server`, `artemis-jms-client` and `artemis-amqp-protocol` 2.19.1 (2.20.0 needs Java 11), and `qpid-jms-client` 0.61.0 (1.0.0 needs Java 11). ## 8. Tests - **Unit and integration tests with in-JVM brokers, no Docker.** The same source tests run against three providers, to show the connector is not tied to one: ActiveMQ Classic over OpenWire, Artemis over its core protocol, and Artemis over AMQP 1.0 through Qpid JMS. - **Cases:** - nothing is committed before checkpoint completion; - messages received after a snapshot are not committed by that checkpoint; - an aborted checkpoint followed by a completed one commits both epochs; - the final checkpoint's epoch is committed after `pollNext` has stopped; - a sealed session can be committed after its consumer is closed, and from a thread other than the one that received the messages; - a repeated completion notification is harmless; - a closed reader's uncommitted messages are redelivered; - the uncommitted-message bound pauses receiving; - parallel readers; - cancellation during connect. - No `Thread.sleep` waits. - **E2E:** a `connector-jms-e2e` module against an `apache/activemq-artemis` container, on Zeta and, if the translation layer supports it, Flink. ## 9. Deliverables - The `connector-jms` module. - `plugin-mapping.properties` entries (`seatunnel.source.Jms` and `seatunnel.sink.Jms` = `connector-jms`). - Registration in `seatunnel-dist`, `plugin_config` and the CI label scope. - EN and ZH docs with vendor examples. - Unit tests and the E2E module. ## Out of scope for v1 Exactly once and XA; topics, including durable and shared subscriptions; `jakarta.jms` and JMS 2.0 APIs; `MapMessage`, `StreamMessage` and `ObjectMessage`; message headers and properties as columns; bounded mode; Spark source; vendor-specific options beyond the factory properties. ## Questions for maintainers 1. Is the scope in section 1 the right scope for this row? 2. Are the plugin name `Jms` and the module name `connector-jms` acceptable? 3. Is `plugins/connector-jms/` the preferred place for vendor client jars? And should the connector bundle the Geronimo JMS 1.1 API jar, or rely on the API jar each vendor client already brings (section 7)? 4. Source first, then sink: is that split acceptable? I'd like to implement this. Please assign this issue to me. I am also claiming the row in #10753, and I will start coding once these questions are answered. ### Usage Scenario Many enterprises move data through JMS brokers such as IBM MQ, TIBCO EMS, Solace and ActiveMQ Artemis, often between core systems that will not move to Kafka. Among JMS brokers, SeaTunnel has a source only for ActiveMQ Classic (#12341). A generic JMS connector would let these users read queue messages into SeaTunnel, without loss across restarts, and write to databases, lakes or Kafka, and publish SeaTunnel output back to the same brokers, without a connector per vendor. ### Related issues - #10753: the tracker row this addresses. - #12341: the ActiveMQ Classic source this builds on and differs from. - #8581: the earlier ActiveMQ source, and the `pollNext` contract from its review. - #12883 / #12910: ActiveMQ source options in review. This proposal does not touch `connector-activemq`. ### Are you willing to submit a PR? - [X] Yes I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
