feat(connectors): add generic JDBC source connector#3588
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report✅ All modified and coverable lines are covered by tests. 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
🚀 New features to boost your workflow:
|
80435e8 to
9e0c750
Compare
| query = query.replace("{last_offset}", "e_sql_literal(offset)); | ||
| } else if let Some(ref initial) = self.config.initial_offset { | ||
| query = query.replace("{last_offset}", "e_sql_literal(initial)); | ||
| } else { |
There was a problem hiding this comment.
{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.
| } | ||
|
|
||
| /// Create HikariCP connection pool | ||
| fn create_connection_pool_internal(&self, env: &mut JNIEnv) -> Result<GlobalRef, Error> { |
There was a problem hiding this comment.
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.
| let state = self.state.lock().expect("state mutex poisoned"); | ||
| self.build_query(&state) | ||
| }; | ||
| info!("Executing query: {}", query); |
There was a problem hiding this comment.
info!("Executing query: {}", query) logs substituted offset/filters. Fix: debug! or redact.
|
/author |
github-actions
Bot
added
S-waiting-on-author
Sure, closed the parent PR with a note mentioning two different PR approach for source and sink. |
9966922 to
6d7c88b
Compare
|
/ready |
github-actions
Bot
added
S-waiting-on-review
| } | ||
|
|
||
| /// Best-effort `Connection.isValid(timeout)` check. Returns false on any | ||
| /// JNI error so the caller re-establishes the connection. |
There was a problem hiding this comment.
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.
| /// | ||
| /// 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> { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
| Ok(s) => s, | ||
| Err(_) => return Err(classify_query_failure(env, "prepare statement")), | ||
| }; | ||
|
|
There was a problem hiding this comment.
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.
| info!("JDBC source connector [{}] opened successfully", self.id); | ||
| Ok(()) | ||
| } | ||
|
|
There was a problem hiding this comment.
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().
|
|
||
| /// 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). |
There was a problem hiding this comment.
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.
|
|
||
| /// 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. |
There was a problem hiding this comment.
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.
| } | ||
| }; | ||
|
|
||
| let columns = self.read_column_metadata(env, &result_set)?; |
There was a problem hiding this comment.
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.
| "Creating direct JDBC connection to: {}", | ||
| sanitize_jdbc_url(self.config.jdbc_url.expose_secret()) | ||
| ); | ||
| let conn = self.create_direct_connection_internal(&mut env)?; |
There was a problem hiding this comment.
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.
|
|
||
| #[async_trait] | ||
| impl Source for JdbcSource { | ||
| async fn open(&mut self) -> Result<(), Error> { |
There was a problem hiding this comment.
lib.rs:1009-1026 — open() never dry-runs build_query(), bad config only surfaces after the first poll_interval sleep. Fix: call build_query() once inside open() before returning Ok.
|
|
||
| // Without an offset yet, drop the incremental predicate but keep the rest | ||
| // of the query (e.g. an ORDER BY) intact. | ||
| if offset.is_none() { |
There was a problem hiding this comment.
lib.rs:755 — WHERE-auto-remove matches only an exact literal substring, brittle to casing/whitespace. Fix: replace the literal match with a case/whitespace-tolerant regex.
| .map_err(|e| Error::InitError(format!("Failed to create JDBC URL string: {}", e)))?; | ||
|
|
||
| // If username/password are provided separately, use 3-arg getConnection | ||
| let connection = if let (Some(username), Some(password)) = |
There was a problem hiding this comment.
lib.rs:385-386 — half-set username/password silently falls through to embedded-URL auth. Fix: validate in open() that username and password are both set or both unset, error otherwise.
| use std::time::Duration; | ||
| use tracing::{debug, info}; | ||
| use uuid::Uuid; | ||
|
|
There was a problem hiding this comment.
local secret serde modules bypass the repo's shared iggy_common::serde_secret helper, crate doesn't even depend on iggy_common. Fix: add the iggy_common dependency and use
serialize_secret/serialize_optional_secret (verify exact names against core/common/src/utils/serde_secret.rs), drop the local serialize halves.
| .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '.') | ||
| } | ||
|
|
||
| /// Classify a JDBC `SQLState` class (first 2 chars) as transient vs permanent. |
There was a problem hiding this comment.
1194-1220 vs sdk/src/source.rs:182-198 — SQLState transient/permanent classification has zero observable runtime effect. Fix: needs SDK-side change to branch on Error variant and add backoff in the poll loop; if out of scope for this PR, remove the "runtime backs off appropriately" claim from the PR description/comments until it's true.
|
|
||
| /// Acquire the direct connection, transparently re-establishing it if it has | ||
| /// dropped since the last poll. | ||
| fn get_connection<'local>(&self, env: &mut JNIEnv<'local>) -> Result<JObject<'local>, Error> { |
There was a problem hiding this comment.
get_connection TOCTOU shape, not reachable today but relies on undocumented single-caller discipline. Fix: collapse the check/close/create/store sequence into one lock scope so
correctness doesn't depend on external call guarantees.
| } | ||
|
|
||
| /// Create a direct JDBC connection via DriverManager | ||
| fn create_direct_connection_internal(&self, env: &mut JNIEnv) -> Result<GlobalRef, Error> { |
There was a problem hiding this comment.
[nit] lib.rs:328-435 — create_direct_connection_internal missing local-frame management used elsewhere in the file. Fix: wrap in push_local_frame/pop_local_frame like read_column_metadata/read_rows do.
| result_set: &JObject, | ||
| columns: &[(String, i32)], | ||
| ) -> Result<(Vec<ProducedMessage>, u64, Option<String>), Error> { | ||
| let mut messages = Vec::new(); |
There was a problem hiding this comment.
[nit] lib.rs:630 — Vec::new() despite known upper bound. Fix: Vec::with_capacity(self.config.batch_size as usize).
| } | ||
|
|
||
| async fn poll(&self) -> Result<ProducedMessages, Error> { | ||
| // Sleep for poll interval |
There was a problem hiding this comment.
[nit] lib.rs:1029-1030 — sleep-before-query causes cadence drift and delays the first poll a full interval. Fix: measure elapsed work time and sleep only the remainder of poll_interval.
| .map_err(|e| Error::Serialization(format!("Failed to serialize row data: {}", e)))? | ||
| }; | ||
|
|
||
| Ok(ProducedMessage { |
There was a problem hiding this comment.
[nit] lib.rs:727-734 — Utc::now() called twice per message. Fix: call once, reuse for both timestamp fields.
| let col_idx = (idx + 1) as i32; | ||
| let value = self.extract_column_value(env, result_set, col_idx, col_type)?; | ||
|
|
||
| let final_col_name = if self.config.snake_case_columns { |
There was a problem hiding this comment.
[nit] lib.rs:689-703 — snake_case_columns collisions silently overwrite, no detection. Fix: warn when a computed snake_case name collides with an existing key in the row map.
| pub driver_class: String, | ||
|
|
||
| /// Path to JDBC driver JAR file | ||
| pub driver_jar_path: String, |
There was a problem hiding this comment.
[nit] lib.rs:91 — driver_jar_path not existence-checked before JVM start, failure surfaces as an opaque wrapped exception. Fix: check the path exists in initialize_jvm/open() and return a direct, actionable error first. then
add -Djava.class.path= — in the README.
| let username_jstring = env.new_string(username).map_err(|e| { | ||
| Error::InitError(format!("Failed to create username string: {}", e)) | ||
| })?; | ||
| let password_jstring = env.new_string(password.expose_secret()).map_err(|e| { |
There was a problem hiding this comment.
[nit] lib.rs:392 — password reaches the JVM heap as an unzeroed java.lang.String, defeats SecretString zeroing. Fix: none available (inherent to the JDBC API surface) — document as an accepted risk in the
README.
| # KIND, either express or implied. See the License for the | ||
| # specific language governing permissions and limitations | ||
| # under the License. | ||
|
|
There was a problem hiding this comment.
[nit] README/config.toml — no credential-precedence note for username/password vs URL-embedded credentials. Fix: add a short "Credential precedence" paragraph to the README.
| // KIND, either express or implied. See the License for the | ||
| // specific language governing permissions and limitations | ||
| // under the License. | ||
|
|
There was a problem hiding this comment.
[nit] test_with_postgres.rs naming — diverges from sibling postgres_source.rs convention. Fix: rename tests to match the sibling's shorter predicate-style naming.
|
Good progress. There are similar pattern of issues are repeated. Think about using common methods where ever possible. |
|
/author |
github-actions
Bot
added
S-waiting-on-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.
|
Thanks for the thorough review. I have pushed a revision that addresses the feedback. |
|
/ready |
github-actions
Bot
added
S-waiting-on-review
3 participants
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.sqlAPI 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
core/connectors/sources/jdbc_source/src/lib.rsopen()(JVM init, driver load, connection) thenpoll()thenbuild_query(offset substitution viaquote_sql_literal) thenread_rows/extract_column_value(JDBC to JSON) thenbuild_message(offset state persistence)get_or_create_jvm,get_connection,read_rowspush_local_frame/pop_local_frameso large result sets do not overflow the JNI localisValidreconnect, pooled-connection returnextract_column_valueNUMERIC/DECIMALas string, binary as base64,BIGINTcaveat documented in the READMEis_transient_sql_state,classify_query_failure,take_pending_sql_exception08/40/53/57/58are transient; pending Java exceptions are clearedDebugimpl, the secret serde modules,sanitize_jdbc_urljdbc_urlandpasswordareSecretString, never logged or serializedcore/integration/tests/connectors/mod.rssetup_runtimeand theConnectorsRuntime/ConnectorsIggyClienthelpers used by the JDBC (and future connector) testscore/server/src/tcp/tcp_listener.rs(3 lines)PortReserver) and waits for the server to writecurrent_config.tomlto learnport == 0), so a fixed-port server hung. The fix always notifies the config writer once the listener binds. TheSO_REUSEPORTcross-shard broadcast stays gated onport == 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, itsconfig.toml, andcore/connectors/runtime/example_config/connectors/jdbc_*.toml.Local Execution
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), andmarkdownlint.test time is about 20 to 25 seconds; unit tests are effectively instant.
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.
prekwas not run (not installed locally); the equivalent checks above were run manually.Limitations
AI Usage