feat(connectors): add generic JDBC source and sink connectors (#2500)

[shbhmrzd]

Which issue does this PR address?

Closes #2500

Rationale

Iggy only shipped PostgreSQL-specific connectors, so every other database needed a purpose-built connector. A generic JDBC pair covers any JDBC-compliant database (PostgreSQL, MySQL, Oracle, SQL Server, H2, and others) through their existing JDBC drivers, instead of writing and maintaining a new connector per database.

What changed?

Before: reading from or writing to anything other than PostgreSQL required a bespoke connector. After: two new connectors talk to any JDBC database through an embedded JVM over JNI, reusing the standard java.sql API and the database's own driver JAR.

The source polls a configured SQL query and produces each row as a JSON message, in bulk (full re-read) or incremental mode (tracks a {last_offset} placeholder and persists the offset across restarts). The sink consumes messages and INSERTs each into a target table, with optional auto-create-table and Iggy metadata columns. Both share one embedded-JVM/JNI bridge, so a single driver JAR serves either direction.

How to review

Area	Where to look	What to check
Source logic	core/connectors/sources/jdbc_source/src/lib.rs	poll() then build_query (offset substitution via quote_sql_literal) then read_rows / extract_column_value (JDBC to JSON type mapping) then build_message (offset state persistence)
Sink logic	core/connectors/sinks/jdbc_sink/src/lib.rs	open() (JVM, connectivity SELECT 1, optional auto-create-table) then consume() then write_messages / insert_batch (batched PreparedStatement) then bind_row
JVM and JNI lifetime	get_or_create_jvm, connection acquire/close in both crates	one JVM per process (shared across instances), per-row and per-message push_local_frame / pop_local_frame, pooled-connection return, isValid reconnect in the source
Type fidelity	source extract_column_value	NUMERIC and DECIMAL map to string (no f64 precision loss), binary maps to base64, BIGINT caveat documented
Error handling	is_transient_sql_state and the classify helpers in both crates	SQLState classes 08, 40, 53, 57, 58 are transient (retried), everything else is permanent. The sink skips and counts permanent batches, retries transient ones
Secrets	config Debug impls, the secret serde modules, sanitize_jdbc_url	URL and password are SecretString, never logged or serialized
Shared test harness	core/integration/tests/connectors/mod.rs	setup_runtime, plus send_messages / get_messages helpers used by the sink and source tests
Server change	core/server/src/tcp/tcp_listener.rs (3 lines)	The connectors integration harness reserves a fixed TCP port (PortReserver) and waits for the server to write current_config.toml to learn the bound address. The server previously wrote that file only when the port was dynamic (port == 0), so a fixed-port server hung. The fix always notifies the config writer once the listener binds. The SO_REUSEPORT cross-shard broadcast stays gated on port == 0, so multi-shard behaviour is unchanged. Worth a check for cluster mode.

Config reference and per-database examples live in each connector's README.md, config.toml, and the core/connectors/runtime/example_config/connectors/jdbc_*.toml files.

Local Execution

• Passed. Ran the CI-equivalent gates locally: cargo fmt --all -- --check, cargo clippy --all-targets --all-features -- -D warnings, cargo check --all --all-features, cargo sort --check, cargo machete, license headers (hawkeye check), and markdownlint.
• Tests: 33 source and 22 sink unit tests, plus 7 Dockerized Postgres integration tests (source basic, multi-row, metadata, incremental offset advance, large result set; sink single and multi-row).
• Validated live end to end: ran iggy-server plus separate source and sink connector runtimes against a Postgres container and pushed 150 messages through source then iggy then sink, confirming all 150 rows arrived intact in the destination table with contiguous offsets and exact decimal values.
• Pre-commit hooks: prek was not run (not installed locally); the equivalent checks listed above (fmt, clippy, sort, machete, license headers, markdownlint, tests) were run manually instead.

Limitations

• JNI permits one JVM per process, so a JDBC source and a JDBC sink cannot run in the same connectors-runtime process. Run them in separate runtime processes. Documented in the connector READMEs.
• JDBC calls are synchronous (blocking) on the runtime worker thread, consistent with the JDBC source model.
• The connectors require a JVM and a JDBC driver JAR at runtime (unlike the pure-Rust connectors). They are added to the release plugin list with that caveat.

AI Usage

1. Tools: Claude Code (Claude).
2. Scope: used substantially for drafting the connector implementations, tests, docs, and example configs, and for an iterative review and hardening pass.
3. Verification: unit tests, Dockerized Postgres integration tests (including incremental-offset and large-result-set cases), and a manual end-to-end run of the full source to iggy to sink pipeline with 150 messages, inspecting the resulting database rows.
4. Can explain every line: yes.

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[Standing-Man]

Hi @shbhmrzd, If possible, please split the Sink and Source changes into two separate PRs so reviewers can review them more effectively. cc @hubcio.

[hubcio]

agreed @Standing-Man. question @shbhmrzd - how much time time does it take for your tests to finish, on your machine? only the ones that you added.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.03%. Comparing base (307fdb1) to head (02bcea7).

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3576      +/-   ##
============================================
- Coverage     74.07%   74.03%   -0.04%     
  Complexity      937      937              
============================================
  Files          1249     1249              
  Lines        128248   128246       -2     
  Branches     104116   104159      +43     
============================================
- Hits          94994    94952      -42     
+ Misses        30219    30214       -5     
- Partials       3035     3080      +45     

Components	Coverage Δ
Rust Core	74.71% <ø> (+<0.01%)	⬆️
Java SDK	62.44% <ø> (ø)
C# SDK	71.40% <ø> (-0.66%)	⬇️
Python SDK	88.88% <ø> (ø)
PHP SDK	84.29% <ø> (ø)
Node SDK	91.35% <ø> (ø)
Go SDK	40.14% <ø> (ø)

Files with missing lines	Coverage Δ
core/server/src/tcp/tcp_listener.rs	75.00% <ø> (+0.49%)	⬆️

... and 17 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[shbhmrzd]

agreed @Standing-Man. question @shbhmrzd - how much time time does it take for your tests to finish, on your machine? only the ones that you added.

Hey @Standing-Man @hubcio
Sure, let me close this and split it into a separate source and sink PRs. I will start of with the source PR. Since the sink code relies on shared utils, I can share sink PR once source PR is merged in.

For the tests, the first run on a fresh machine is a bit longer because it does a one-time pull of the Postgres image and a ~1MB JDBC driver download, but the subsequent runs finish in under a minute.