feat(targets): add configurable batch wait time limit - #3757
Draft
imamAtif wants to merge 6 commits into
Draft
Conversation
Add a new target-level setting `batch_wait_limit_seconds` that allows users to set a maximum time a batch can remain open before being processed, regardless of how many rows it contains. This addresses the case where a target runs in a memory-constrained environment and cannot afford to wait for large batches to fill up. - Add `TARGET_BATCH_WAIT_LIMIT_SECONDS_CONFIG` to built-in config - Track batch start time in `Sink._batch_start_time` - Add `Sink.is_too_old` property to check elapsed time - Update `Sink.is_full` to trigger drain when time limit is exceeded - Reset batch start time in `Sink.mark_drained()` Closes meltano#1626
Contributor
Reviewer's GuideIntroduces the optional Sequence diagram for time-based batch drainingsequenceDiagram
participant Target
participant Sink
Target->>Sink: _after_process_record(context)
alt first record in batch
Sink->>Sink: _batch_start_time = time.time()
end
Target->>Sink: is_full
Sink->>Sink: current_size >= max_size
Sink->>Sink: is_too_old
Sink->>Sink: time.time() - _batch_start_time >= batch_wait_limit_seconds
alt row limit or wait limit reached
Target->>Sink: mark_drained()
Sink->>Sink: _batch_start_time = None
end
File-Level Changes
Assessment against linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Documentation build overview
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="singer_sdk/sinks/core.py" line_range="335" />
<code_context>
def is_full(self) -> bool:
"""True if the sink needs to be drained."""
- return self.current_size >= self.max_size
+ return self.current_size >= self.max_size or self.is_too_old
@property
</code_context>
<issue_to_address>
**issue (bug_risk):** The time limit is checked only after a record has been processed, and the batch timer is started only in `_after_process_record`; a batch that has already exceeded its limit is not drained until another record arrives, while time spent processing the first record is excluded from the limit. A low-volume stream therefore keeps an open batch beyond the configured maximum.
**Triggers:** When no record arrives after the batch exceeds its time limit, or when processing the first record itself takes longer than the configured limit.
**Suggested fix:** Start the timer before processing the first record and add a timer-driven or end-of-input drain path so expired batches do not depend on a subsequent record.
</issue_to_address>
### Comment 2
<location path="singer_sdk/helpers/capabilities.py" line_range="308-315" />
<code_context>
),
).to_dict()
+TARGET_BATCH_WAIT_LIMIT_SECONDS_CONFIG = PropertiesList(
+ Property(
+ "batch_wait_limit_seconds",
+ IntegerType,
+ title="Batch Wait Limit (Seconds)",
+ description=(
+ "Maximum number of seconds to wait for a batch to reach "
+ "the configured batch size before processing it anyway."
+ ),
+ ),
+).to_dict()
</code_context>
<issue_to_address>
**issue (bug_risk):** The new setting accepts arbitrary integers because `IntegerType` has no non-negative constraint, so a negative value passes config validation and makes `is_too_old` true immediately after the first record, forcing every batch to drain after each record.
**Triggers:** When a user configures a negative `batch_wait_limit_seconds` value.
**Suggested fix:** Constrain the schema to non-negative values, or reject values below zero when the sink is initialized.
</issue_to_address>
### Comment 3
<location path="singer_sdk/sinks/core.py" line_range="358" />
<code_context>
+ """True if the current batch has exceeded the wait time limit."""
+ if self._batch_start_time is None or self.batch_wait_limit_seconds is None:
+ return False
+ return (time.time() - self._batch_start_time) >= self.batch_wait_limit_seconds
+
@property
</code_context>
<issue_to_address>
**issue (bug_risk):** Elapsed time is calculated with `time.time()`, which is wall-clock time; a system clock adjustment backward makes the elapsed duration negative and delays draining beyond the configured limit, while a forward adjustment causes premature draining.
**Triggers:** When the system clock is corrected or synchronized while a batch is open.
**Suggested fix:** Use `time.monotonic()` for batch elapsed-time measurements and initialize/reset the timer from the same monotonic clock.
</issue_to_address>Sourcery assessment
Approval pending. 3 findings to address first.
Blocking findings: singer_sdk/sinks/core.py:335, singer_sdk/helpers/capabilities.py:315, singer_sdk/sinks/core.py:358
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #3757 +/- ##
==========================================
+ Coverage 94.35% 94.42% +0.06%
==========================================
Files 74 74
Lines 6294 6314 +20
Branches 770 775 +5
==========================================
+ Hits 5939 5962 +23
+ Misses 266 264 -2
+ Partials 89 88 -1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
- use time.monotonic() instead of time.time() to avoid issues with system clock adjustments - clamp negative batch_wait_limit_seconds to None (disabled) instead of allowing invalid values through - move batch timer start from _after_process_record to preprocess_record so it starts before the first record is processed, not after
Check all active sinks for expired batches at the start of _process_record_message, so batches that exceeded their wait time limit while idle are drained before processing the next record.
Add tests for expired batch drain flow and preprocess_record timer initialization to improve codecov patch coverage.
Update the expired batch test to go through _process_record_message with a properly registered stream, covering the drain loop at the start of record processing.
Author
|
All three concerns addressed in subsequent commits:
|
imamAtif
marked this pull request as draft
September 4, 2026 15:40
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
batch_wait_limit_secondsthat sets the maximum time a batch can remain open before being processed, regardless of row countSink._batch_start_timeand reset it after each drainSink.is_too_oldproperty to check elapsed time against the configured limitSink.is_fullto trigger a drain when the time limit is exceededThis addresses the case where a target runs in a memory-constrained environment and cannot afford to wait for large batches to fill up. The existing
batch_size_rowssetting controls the row count; this adds the complementary time-based control.Validation
uv run pytest -q— 859 passed, 388 deselected, 1 xfailed, 21 subtests passedFiles changed
singer_sdk/helpers/capabilities.py— newTARGET_BATCH_WAIT_LIMIT_SECONDS_CONFIGsinger_sdk/sinks/core.py—batch_wait_limit_seconds,_batch_start_time,is_too_old, updatedis_fullandmark_drainedsinger_sdk/target_base.py— import and merge new config inappend_builtin_config, drain expired sinks at start of record processingtests/core/test_target_base.py— 10 new tests for config,is_too_old,is_fullintegration, drain reset, about info, negative values, expired batch drain, timer inittests/sql/test_target.py— addedbatch_wait_limit_secondsto expected default settingsCloses #1626
Summary by Sourcery
Add time-based batch draining for targets to prevent low-volume batches from remaining open indefinitely.
New Features:
Bug Fixes:
Enhancements:
Tests: