chovy-3012 opened a new pull request, #12910: URL: https://github.com/apache/seatunnel/pull/12910
### Purpose of this pull request Acknowledging contribution to Apache SeaTunnel. This PR resolves issue #12883. It adds `message_selector`, `receive_timeout`, and seven connection factory tuning options to the ActiveMQ source connector (previously hardcoded or unavailable). It also extracts `ActivemqBaseOptions extends ConnectorCommonOptions` — 12 shared options that both source and sink now inherit. #### Options **Shared — `ActivemqBaseOptions` (source + sink):** | Option | Type | Default | Description | |--------|------|---------|-------------| | `uri` | string | — | Broker address (`tcp://host:port`, `ssl://host:port`) | | `username` | string | — | Broker username; must pair with `password` | | `password` | string | — | Broker password; must pair with `username` | | `queue_name` | string | — | Queue name | | `client_id` | string | — | JMS clientID for the connection | | `check_for_duplicate` | boolean | — | Consumer duplicate message detection | | `always_session_async` | boolean | — | Separate thread per session for dispatch | | `close_timeout` | int | 5000 | Close timeout (ms). Source applies default via `config.get()`; sink only when explicitly set | | `connect_response_timeout` | int | 10000 | Connection response timeout (ms). Source applies default; sink only when explicitly set | | `dispatch_async` | boolean | — | Broker async dispatch to consumer | | `warn_about_unstarted_connection_timeout` | int | — | Warning timeout for unstarted connection; <0 disables | | `consumer_expiry_check_enabled` | boolean | — | Message expiration check before dispatch | **Source-specific — `ActivemqSourceOptions`:** | Option | Type | Default | Description | |--------|------|---------|-------------| | `format` | enum (JSON/TEXT) | JSON | Message deserialization format | | `field_delimiter` | string | `,` | Field delimiter for TEXT format | | `max_in_flight_messages` | int | 1000 | Max unacknowledged messages per reader; also sets queue prefetch | | `message_selector` | string | — | JMS selector for broker-side filtering | | `receive_timeout` | int | 200 | `consumer.receive()` timeout (ms) | **Sink-specific — `ActivemqSinkOptions`:** | Option | Type | Default | Description | |--------|------|---------|-------------| | `always_sync_send` | boolean | — | Always use synchronous sends | | `nested_map_and_list_enabled` | boolean | — | Nested Map/List in message properties | | `max_thread_pool_size` | int | 1000 | Connection factory thread pool size | | `send_timeout` | int | 0 | Send timeout (ms); 0 = no timeout | | `use_compression` | boolean | false | Compress message bodies | | `delivery_mode` | int | 2 | JMS delivery mode (1=NON_PERSISTENT, 2=PERSISTENT) | | `priority` | int | 4 | JMS message priority (0–9) | | `producer_window_size` | int | 0 | Flow control window for async sends; 0 = none | | `time_to_live` | long | 0 | Message TTL (ms); 0 = never expires | | `use_async_send` | boolean | false | Send without waiting for broker confirmation | ### Does this PR introduce _any_ user-facing change? Yes. Nine new user-configurable options on the source. Two option defaults changed: - `close_timeout`: `noDefaultValue()` → `defaultValue(5000)`. Source behavior unchanged (was hardcoded 5000). Sink behavior unchanged — uses `getOptional().ifPresent()` to preserve the ActiveMQ client default. - `connect_response_timeout`: `defaultValue(0)` → `defaultValue(10000)`. Source behavior unchanged (was hardcoded 10000). Sink minor change: previously explicitly set to 0 when unset (overwriting `jms.*` URI params); now preserves the ActiveMQ client default and `jms.*` URI params. Effective runtime default is the same (0). Docs updated in `docs/en` and `docs/zh` for both source and sink. ### How was this patch tested? - 57/57 unit tests pass (11 in `ActivemqFactoryTest`, 33 in `ActivemqSourceReaderTest`, plus sink tests). - 9 E2E tests in `ActivemqIT`, including `testSourceMessageSelectorFiltersMessages` (publishes car + truck messages, selector `msg_type = 'car'`, verifies only matches consumed). - Spotless clean. ### Check list * [x] If necessary, please update the documentation to describe the new feature. * [x] If necessary, please update `incompatible-changes.md` to describe the incompatibility caused by this PR. * [x] Add e2e testcase in `seatunnel-e2e`. * [ ] Update `plugin-mapping.properties` — no change needed (connector already registered). * [ ] Update the pom file of `seatunnel-dist` — no change needed (no new module). * [ ] Add ci label in `label-scope-conf` — no change needed (connector already labeled). * [ ] Update connector `plugin_config` — no change needed (no new jar). -- 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]
