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 #4225 +/- ##
============================================
- Coverage 87.13% 85.63% -1.51%
Complexity 1455 1455
============================================
Files 1273 1273
Lines 215301 216032 +731
Branches 179911 180028 +117
============================================
- Hits 187611 184996 -2615
- Misses 23173 26566 +3393
+ Partials 4517 4470 -47
🚀 New features to boost your workflow:
|
|
@Luishfs Seems your cargo.lock file is stale, due to which 11/13 checks are failing. The remaining two checks that fail are:
|
hubcio
added
S-waiting-on-author
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
left line comments for the cursor handling, the unbounded first poll and the crate metadata. one thing that doesn't sit on a diff line: core/connectors/sources/README.md still lists five sources in the Available Sources table, and mongodb_source is the only source directory without a README of its own. worth adding the row and a plugin README that covers query and max_pool_size.
| }; | ||
| messages.push(message); | ||
| } | ||
| let mut state = self.state.lock().await; |
There was a problem hiding this comment.
critical: search_documents writes state.last_poll_timestamp inside poll, but on_batch_result keeps the default no-op, so a runtime Nack re-polls past the dropped batch. stage the cursor in poll and commit it on Ack only, as random_source does.
| && let Some(last_timestamp) = state.last_poll_timestamp | ||
| { | ||
| let mut filter = self.config.query.clone().unwrap_or_default(); | ||
| filter.insert( |
There was a problem hiding this comment.
critical: the cursor stores the newest timestamp of a batch that limit truncated, so documents sharing that millisecond beyond the cutoff never come back. keep the last _id as a tiebreaker, or use $gte with a skip count.
| .await | ||
| .map_err(|e| Error::InitError(format!("Failed to move cursor {e}")))? | ||
| { | ||
| if let Some(timestamp_field) = &self.config.timestamp_field |
There was a problem hiding this comment.
critical: as_datetime() returns None for a string, an int, or a missing field, so last_poll_timestamp never advances and every poll takes the unbounded branch. check the field type in open, or parse strings and numbers.
| .map_err(|e| Error::InitError(format!("Failed to execute search: {e}")))? | ||
| } else { | ||
| let filter = &self.config.query.clone().unwrap_or_else(|| doc! {}); | ||
| coll.find(filter.clone()) |
There was a problem hiding this comment.
critical: the else branch runs coll.find(filter) with no .limit(), so a first poll on a large collection loads all of it into one batch. apply the same limit and sort on both branches.
| "core/connectors/sources/elasticsearch_source", | ||
| "core/connectors/sources/http_source", | ||
| "core/connectors/sources/influxdb_source", | ||
| "core/connectors/sources/mongodb_source", |
There was a problem hiding this comment.
critical: this adds the workspace member and leaves Cargo.lock stale, so every --locked build fails at this head. commit the regenerated Cargo.lock.
| .as_ref() | ||
| .ok_or_else(|| Error::Storage("Mongodb client not initialized".to_string()))?; | ||
|
|
||
| let messages = match self.search_documents(client).await { |
There was a problem hiding this comment.
simplification: this match passes the error straight through, so the Err arm adds nothing. replace it with let messages = self.search_documents(client).await?;.
| } | ||
|
|
||
| async fn search_documents(&self, client: &Client) -> Result<Vec<ProducedMessage>, Error> { | ||
| let state = self.state.lock().await; |
There was a problem hiding this comment.
simplification: the state guard is held across find only because the second lock at line 180 forces the manual drop. read last_poll_timestamp into a local first, then the drop and its import both go away.
also at lines 19, 145.
| doc! { "$gt": mongodb::bson::DateTime::from_millis(last_timestamp.timestamp_millis()) }, | ||
| ); | ||
|
|
||
| coll.find(filter.clone()) |
There was a problem hiding this comment.
simplification: filter is cloned here and at 140 although find takes it by value, and lines 121-122 build String names that the driver takes as &str. pass the filter by value and borrow the names.
| readme = "../../README.md" | ||
|
|
||
| [package.metadata.cargo-machete] | ||
| ignored = ["dashmap", "once_cell", "simd-json"] |
There was a problem hiding this comment.
simplification: dashmap and simd-json are declared but never named, and the machete ignore list hides them along with a stale once_cell entry. delete both dependencies and the ignore block.
| } | ||
|
|
||
| impl MongodbSourceFixture { | ||
| #[allow(dead_code)] |
There was a problem hiding this comment.
simplification: database_name() and collection_name() have no callers, and the two #[allow(dead_code)] attributes hide that. delete both.
3 participants
Which issue does this PR address?
Status: Open.
#3042
Closes #
Relates to #
Rationale
Adding MongoDB source to connectors
What changed?
Added mongoDB source
Local Execution
Passed. Video showing MongoDB Source behaviour:
simplescreenrecorder3-2026-09-18_13.54.55.mp4
also ran all pre-commit hooks and nothing seems related to my changes:
AI Usage
Claude
2.Scope of usage?
Mostly for small corrections or questions
3.How did you verify the generated code works correctly?
N/A
4.Can you explain every line of the code if asked?
its been some time but yeah =P