feat(connectors): add generic JDBC source connector by shbhmrzd · Pull Request #3588 · apache/iggy

shbhmrzd · 2026-07-01T04:50:15Z

Which issue does this PR address?

Relates to #2500

(This is the first of two PRs splitting #2500 for easier review: the JDBC source connector here, and the JDBC sink connector as a follow-up PR once this lands. This PR also carries the shared connector test harness and a small server-side fix that both connectors' integration tests rely on.)

Rationale

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

What changed?

Adds a JDBC source connector that talks to any JDBC database through an embedded JVM over JNI, using the standard java.sql API and the database's own driver JAR. It polls a configured SQL query and produces each row as a JSON message, in bulk mode (re-reads the query result each poll, capped at batch_size) or incremental mode (tracks a {last_offset} placeholder and persists the offset across restarts). Column values are mapped to JSON with type fidelity in mind (decimals as strings to avoid f64 precision loss, binary as base64), and query failures are classified into transient vs permanent error variants by SQLState, which sets the returned error variant and the log message. It does not currently change the connector's retry or backoff behavior.

How to review

Area	Where to look	What to check
Source logic	`core/connectors/sources/jdbc_source/src/lib.rs`	`open()` (JVM init, driver load, connection) then `poll()` then `build_query` (offset substitution via `quote_sql_literal`) then `read_rows` /
`extract_column_value` (JDBC to JSON) then `build_message` (offset state persistence)
JVM and JNI lifetime	`get_or_create_jvm`, `get_connection`, `read_rows`	one JVM per process (shared static), per-row `push_local_frame` / `pop_local_frame` so large result sets do not overflow the JNI local
reference table, `isValid` reconnect, pooled-connection return
Type fidelity	`extract_column_value`	`NUMERIC` / `DECIMAL` as string, binary as base64, `BIGINT` caveat documented in the README
Error handling	`is_transient_sql_state`, `classify_query_failure`, `take_pending_sql_exception`	SQLState classes `08` / `40` / `53` / `57` / `58` are transient; pending Java exceptions are cleared
Secrets	config `Debug` impl, the secret serde modules, `sanitize_jdbc_url`	`jdbc_url` and `password` are `SecretString`, never logged or serialized
Shared test harness	`core/integration/tests/connectors/mod.rs`	`setup_runtime` and the `ConnectorsRuntime` / `ConnectorsIggyClient` helpers used by the JDBC (and future connector) 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: core/connectors/sources/jdbc_source/README.md, its config.toml, and core/connectors/runtime/example_config/connectors/jdbc_*.toml.

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 unit tests, plus 5 Dockerized Postgres integration tests (basic, multi-row, metadata, incremental offset advance, and a large-result-set test that exercises the JNI local-frame handling). Integration
test time is about 20 to 25 seconds; unit tests are effectively instant.
Validated live end to end: started iggy-server, created a stream and topic, and ran the JDBC source runtime against a Postgres container seeded with 150 rows. The source read all 150 rows in a single poll,
advanced its incremental offset, and produced exactly 150 messages to the topic with no errors.
Pre-commit hooks: prek was not run (not installed locally); the equivalent checks above were run manually.

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 README.
JDBC calls are synchronous (blocking) on the runtime worker thread.
The connector requires a JVM and a JDBC driver JAR at runtime (unlike the pure-Rust connectors); it is added to the release plugin list with that caveat.

AI Usage

Tools: Claude Code (Claude).
Scope: used substantially for drafting the connector implementation, tests, docs, and example configs, and for an iterative review and hardening pass.
Verification: unit tests, Dockerized Postgres integration tests (including incremental-offset and large-result-set cases), and a manual end-to-end run producing 150 messages from Postgres into iggy.
Can explain every line: yes.

github-actions · 2026-07-01T04:50:27Z

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.

codecov · 2026-07-01T06:27:03Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.11%. Comparing base (d6059a1) to head (9e0c750).
⚠️ Report is 49 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3588      +/-   ##
============================================
- Coverage     74.13%   74.11%   -0.02%     
  Complexity      937      937              
============================================
  Files          1258     1258              
  Lines        131452   131450       -2     
  Branches     107320   107363      +43     
============================================
- Hits          97455    97429      -26     
+ Misses        30909    30892      -17     
- Partials       3088     3129      +41

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

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

... and 19 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.

ryerraguntla · 2026-07-01T18:37:56Z

+                query = query.replace("{last_offset}", &quote_sql_literal(offset));
+            } else if let Some(ref initial) = self.config.initial_offset {
+                query = query.replace("{last_offset}", &quote_sql_literal(initial));
+            } else {


{tracking_column} never substituted in build_query; only {last_offset} replaced. README/examples use WHERE {tracking_column} > {last_offset} → invalid SQL at runtime. Fix: substitute from config.tracking_column or fix docs.

ryerraguntla · 2026-07-01T18:40:33Z

+    }
+
+    /// Create HikariCP connection pool
+    fn create_connection_pool_internal(&self, env: &mut JNIEnv) -> Result<GlobalRef, Error> {


Two places - 472-485,1255 enable_connection_pool loads HikariCP via JNI but JVM classpath = single driver_jar_path only → pool mode fails unless HikariCP bundled. Fix: add HAR to classpath, or disable pool flag until wired.

ryerraguntla · 2026-07-01T18:58:51Z

+            let state = self.state.lock().expect("state mutex poisoned");
+            self.build_query(&state)
+        };
+        info!("Executing query: {}", query);


info!("Executing query: {}", query) logs substituted offset/filters. Fix: debug! or redact.

ryerraguntla · 2026-07-01T19:03:54Z

/author

hubcio

this is described as the first of two PRs splitting #2500, but #3576 - the combined source + sink version - is still open and also closes #2500. is #3576 superseded by this split? if so, closing it would avoid review effort landing on the wrong PR.

shbhmrzd · 2026-07-05T17:37:51Z

this is described as the first of two PRs splitting #2500, but #3576 - the combined source + sink version - is still open and also closes #2500. is #3576 superseded by this split? if so, closing it would avoid review effort landing on the wrong PR.

Sure, closed the parent PR with a note mentioning two different PR approach for source and sink.

shbhmrzd · 2026-07-18T20:28:30Z

/ready

ryerraguntla · 2026-07-19T03:29:21Z

+    }
+
+    /// Best-effort `Connection.isValid(timeout)` check. Returns false on any
+    /// JNI error so the caller re-establishes the connection.


lib.rs:473-478, 551-560,634-988 — JNI exceptions left uncleared on ~45 of 46 fallible call_method/get_string sites (only classify_query_failure clears, 2 sites). Sharpest instance: statement.close() at 557 fires while executeQuery's SQLException is still pending, cleared one line too late. JNI spec forbids further JNI calls while an exception is pending — can abort the whole embedded JVM, i.e. the entire connectors-runtime process, not just this connector. Fix: route every fallible JNI call through an exception-clearing helper.

ryerraguntla · 2026-07-19T03:33:51Z

+    ///
+    /// The mutex is held only briefly: once to read the current offset for
+    /// query building, and once after the JNI work to write the updated state.
+    fn execute_query(&self, env: &mut JNIEnv) -> Result<Vec<ProducedMessage>, Error> {


lib.rs:484-513 vs runtime/src/source.rs:456-517 — state.last_offset advances in-memory as soon as rows are fetched, before delivery is confirmed. Disk-state persist is correctly gated on send
success, but the in-memory offset never rolls back on a transient send failure — that batch is silently and permanently skipped, and the next successful poll's further-advanced offset overwrites disk state,
erasing any replay chance. Breaks the connector's own at-least-once promise on ordinary non-crash failures. Fix: derive next offset from last durably-persisted state, not from a struct field advanced ahead of
confirmed delivery.

This is the same model the existing offset-tracking sources use (postgres_source advances state.tracking_offsets in poll() before delivery too), so the durable guarantee today is at-least-once across restarts but unable to support transient in-process send failure without a restart.

IMO the clean fix can't be done at the connector layer, because the runtime owns the state file and hands the connector its state only once at open(), with no per-poll delivery ack to roll the in-memory offset back. I documented the exact guarantee in the README and would like to track the stronger guarantee as a runtime-level follow-up (roll back / re-feed the offset on send failure), which benefits every offset-tracking source uniformly.
Happy to open that issue. Please let me know your thoughts.

ryerraguntla · 2026-07-19T11:53:06Z

+            Ok(s) => s,
+            Err(_) => return Err(classify_query_failure(env, "prepare statement")),
+        };
+


lib.rs:539-549,743-745 — bulk mode has no pagination beyond setMaxRows; tables larger than batch_size never fully sync, silently. Fix: log a loud warning when a poll returns exactly batch_size rows so truncation is diagnosable; full cross-DB OFFSET pagination is a larger follow-up, scope separately. Also fix the Mode::Bulk doc comment and README, both currently claim "full table scan" unconditionally.

ryerraguntla · 2026-07-19T11:54:09Z

+        info!("JDBC source connector [{}] opened successfully", self.id);
+        Ok(())
+    }
+


lib.rs:1028-1055 — blocking JNI/JDBC I/O runs directly on the shared async runtime; chains with the mutex-poisoning finding since the operator restart path also runs on that same runtime with no timeout of its own. Fix: wrap the fetch body of poll() in tokio::task::block_in_place (works today, no SDK change needed since the runtime is already multi-thread); apply the same wrapping to close().

ryerraguntla · 2026-07-19T11:54:59Z

+
+/// Return the larger of two offset values. Both are compared numerically when
+/// they parse as numbers (covers integer and decimal tracking columns), and
+/// lexicographically otherwise (covers ISO-8601 timestamps and text keys).


lib.rs:1171-1178 — larger_offset compares via f64, loses precision past 2^53, contradicts the file's own NUMERIC/DECIMAL-as-string rationale. Fix: try i128 parse/compare first (exact for the full BIGINT range), fall back to f64 only for genuinely fractional values, lexical for non-numeric.

ryerraguntla · 2026-07-19T11:56:02Z

+
+/// Quote a value as a SQL string literal, escaping embedded single quotes by
+/// doubling them. Used to substitute the incremental `{last_offset}` value,
+/// which originates from a (DB-controlled) tracking-column value.


lib.rs:1148-1150 — quote_sql_literal doesn't escape backslash, MySQL default sql_mode breakout possible (downgraded from critical, requires attacker already has DB write access). Fix: escape backslash
before doubling quotes; real long-term fix is binding the offset as a PreparedStatement parameter instead of string substitution — scope that as a follow-up refactor.

ryerraguntla · 2026-07-19T11:56:54Z

+            }
+        };
+
+        let columns = self.read_column_metadata(env, &result_set)?;


lib.rs:562-569 — statement.close() skipped on read-error path, leaks PreparedStatement/cursor. Fix: split the try-body into its own function and close unconditionally in the caller after it returns, on both Ok and Err.

ryerraguntla · 2026-07-19T12:01:36Z

+            "Creating direct JDBC connection to: {}",
+            sanitize_jdbc_url(self.config.jdbc_url.expose_secret())
+        );
+        let conn = self.create_direct_connection_internal(&mut env)?;


lib.rs:322,441,452,458,461,489,500,1046,1075,1120 — mutex .expect()-on-poison cluster, no self-heal, contradicts repo's no-unwrap/expect rule. Fix: replace .expect() with a helper that maps PoisonError
to a returned Error::Connection/Error::InitError instead of panicking, so a poisoned lock surfaces as a logged failure rather than a permanent panic loop.

ryerraguntla · 2026-07-19T12:24:17Z

/author

shbhmrzd · 2026-07-19T17:07:12Z

Thanks for the thorough review. I have pushed a revision that addresses the feedback.

shbhmrzd · 2026-07-19T17:07:17Z

/ready

ryerraguntla · 2026-07-20T09:27:03Z

+            .or(self.config.initial_offset.as_deref());
+
+        // Without an offset yet, drop the incremental predicate but keep the rest
+        // of the query (e.g. an ORDER BY) intact.


899-900 + :647 + :608 — cold-start incremental drops WHERE (no offset), setMaxRows caps batch, advances max_offset → next poll > max permanently skips unread lower keys when no ORDER BY. Same class any truncated unordered poll. Fix: require ORDER BY {tracking_column} at open (or forbid cold-start without initial_offset + ordered paging). README :208-209 claims order optional = false.

ryerraguntla · 2026-07-20T09:33:36Z

+    /// Validate configuration before touching the JVM or the database, so bad
+    /// config surfaces immediately at `open()` with an actionable message rather
+    /// than as an opaque wrapped JVM error or only after the first poll sleep.
+    fn validate_config(&self) -> Result<(), Error> {


lib.rs:1174-1198 + README :193 — incremental does not require tracking_column → offset never advances → infinite re-read. Fix: fail open() if mode=incremental && no tracking_column; README Required.

ryerraguntla · 2026-07-20T09:34:45Z

+
+        // Update state with results (short lock)
+        {
+            let mut state = lock_mutex(&self.state, "state")?;


lib.rs:608-610 + runtime/src/source.rs:464-515 — in-memory offset before send; disk only after send OK → in-process send fail skips batch. README honest; postgres pattern. Fix: defer mem advance / rollback (follow-up OK).

ryerraguntla · 2026-07-20T09:35:24Z

+        // full batch there is normal paging that the next poll advances past, not
+        // truncation. Full cross-database OFFSET pagination is a separate
+        // follow-up.
+        if self.config.mode == Mode::Bulk && row_count == self.config.batch_size as u64 {


lib.rs:597-603 — bulk setMaxRows truncates; surplus never synced. Fix: paginate or fail-closed.

ryerraguntla · 2026-07-20T09:36:06Z

+        }
+    }
+
+    /// Extract the tracking-column value as a string offset, or `None` when it is


lib.rs:1152-1158 — NULL tracking skipped for watermark; row emitted; offset jumps → never re-fetch. Fix: reject NULL track / IS NOT NULL.

ryerraguntla · 2026-07-20T09:37:22Z

+            row_data.insert(final_col_name.clone(), value);
+
+            // Track offset if this is the tracking column
+            if let Some(ref tracking_col) = self.config.tracking_column


lib.rs:841-844 — offset match raw JDBC col_name not snake final_col_name → miss → dups. Fix: match normalized name.

ryerraguntla · 2026-07-20T09:38:25Z

+
+[package]
+name = "iggy_connector_jdbc_source"
+version = "0.1.0"


Cargo.toml:20,32 + no publish — 0.1.0 / rlib vs siblings 0.4.1-edge.1 / lib + publish=false. Fix: align.

ryerraguntla · 2026-07-20T09:39:20Z

+# Required by source_connector! macro
+dashmap = { workspace = true }
+
+# For parsing duration strings


humantime-serde = "1.1" direct pin; siblings use workspace humantime. Fix: drop; parse like postgres_source.

ryerraguntla · 2026-07-20T09:40:40Z

+Ordering the query by the tracking column is still recommended for readability.
+
+**Example:**
+


README/examples partition_id dead field; Oracle ROWNUM > {last_offset} wrong cursor. Fix: strip / use ID timestamp.

ryerraguntla · 2026-07-20T09:41:47Z

+
+        let now_ms = now.timestamp_millis() as u64;
+        Ok(ProducedMessage {
+            id: Some(Uuid::new_v4().as_u128()),


lib.rs:873 — Uuid::new_v4() → at-least-once not idempotent by msg id. Matches postgres_source. Fix: deterministic id (follow-up).

ryerraguntla · 2026-07-20T09:42:26Z

+        use java::sql::Types;
+
+        // Check if null first
+        let obj = jni!(


double JNI getObject+getter (:989-1017); isValid up to 30s on block_in_place worker (:557). Fix: wasNull+one getter; short liveness timeout.

ryerraguntla · 2026-07-20T09:50:10Z

Still there are few more around the order by clause . Check with what is mentioned in README and the actual implementation.

ryerraguntla · 2026-07-20T09:50:15Z

/author

Adds a JDBC source connector that polls a configured SQL query against any JDBC-compliant database (PostgreSQL, MySQL, Oracle, SQL Server, H2) through an embedded JVM and produces each row as a JSON message. Supports bulk and incremental (offset-tracked) modes with persisted offset state. Also adds the shared connector integration-test harness and a small TCP listener change so that harness can read the server's written runtime config when bound to a fixed port. The matching JDBC sink connector follows in a separate PR.

Aligns the JDBC source crate with the other source connectors: crate version/type and publish flag, and poll_interval parsing via the workspace humantime crate rather than a direct humantime-serde pin. Removes a redundant per-column JNI getObject probe (using wasNull after the typed getter instead) and caps the per-poll liveness check at five seconds so a dead connection cannot stall a shared worker. Drops the unused partition_id field from the example configs and docs.

shbhmrzd · 2026-07-21T05:39:04Z

/ready

github-actions Bot added the S-waiting-on-review PR is waiting on a reviewer label Jul 1, 2026

shbhmrzd mentioned this pull request Jul 1, 2026

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

Closed

shbhmrzd force-pushed the jdbc_source branch 2 times, most recently from 80435e8 to 9e0c750 Compare July 1, 2026 10:22

ryerraguntla reviewed Jul 1, 2026

View reviewed changes

Comment thread core/connectors/sources/jdbc_source/src/lib.rs

ryerraguntla reviewed Jul 1, 2026

View reviewed changes

Comment thread core/connectors/sources/jdbc_source/src/lib.rs

ryerraguntla reviewed Jul 1, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 1, 2026

hubcio requested changes Jul 1, 2026

View reviewed changes

shbhmrzd force-pushed the jdbc_source branch 3 times, most recently from 9966922 to 6d7c88b Compare July 18, 2026 20:16

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Jul 18, 2026

ryerraguntla reviewed Jul 19, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 19, 2026

shbhmrzd force-pushed the jdbc_source branch from 6d7c88b to ae0ea9a Compare July 19, 2026 16:42

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Jul 19, 2026

ryerraguntla reviewed Jul 20, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 20, 2026

shbhmrzd force-pushed the jdbc_source branch from ae0ea9a to e7e1f9b Compare July 20, 2026 15:18

shbhmrzd added 2 commits July 20, 2026 21:50

Merge branch 'master' into jdbc_source

2df58c5

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Jul 21, 2026

		Ordering the query by the tracking column is still recommended for readability.

		Example:

Uh oh!

Conversation

shbhmrzd commented Jul 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR address?

Rationale

What changed?

How to review

Local Execution

Limitations

AI Usage

Uh oh!

github-actions Bot commented Jul 1, 2026

Uh oh!

codecov Bot commented Jul 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryerraguntla commented Jul 1, 2026

Uh oh!

hubcio left a comment

Choose a reason for hiding this comment

Uh oh!

shbhmrzd commented Jul 5, 2026

Uh oh!

shbhmrzd commented Jul 18, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryerraguntla commented Jul 19, 2026

Uh oh!

shbhmrzd commented Jul 19, 2026

Uh oh!

shbhmrzd commented Jul 19, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

shbhmrzd commented Jul 1, 2026 •

edited

Loading

codecov Bot commented Jul 1, 2026 •

edited

Loading