direct_consumer: move offset update logic to fetch_next#28309
Conversation
joe-redpanda
requested review from
bharathv,
Copilot and
michael-redpanda
and removed request for
Copilot
joe-redpanda
changed the title
joe-redpanda
force-pushed
the
start_offset
branch
2 times, most recently
from
08d481a to
75dad5b
Compare
There was a problem hiding this comment.
Pull Request Overview
This PR moves the offset update logic from the fetch response processing stage to the fetch_next method to ensure correctness. The change addresses a timing issue where fetches could become stale between processing and being returned to the consumer. By deferring offset updates until after the final subscription epoch filter is applied in fetch_next, the PR guarantees that only valid, current offsets are stored.
Key Changes:
- Removed premature offset updates in
process_fetch_responsemethod infetcher.cc - Added
update_start_offsetsmethod that updates offsets only after subscription filtering - Refactored subscription lookup logic to use helper methods with
std::reference_wrapper
Reviewed Changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| src/v/kafka/client/direct_consumer/fetcher.cc | Removed premature offset updates and maybe_update_source_partition_offsets call from fetch response processing |
| src/v/kafka/client/direct_consumer/direct_consumer.h | Added new helper methods for subscription lookup and renamed offset update method |
| src/v/kafka/client/direct_consumer/direct_consumer.cc | Implemented update_start_offsets with validation logging and refactored subscription lookup methods |
Retry command for Build#75448please wait until all jobs are finished before running the slash command |
|
The failures here seem related to the change? |
joe-redpanda
force-pushed
the
start_offset
branch
from
75dad5b to
39525c4
Compare
|
/ci-repeat 1 |
joe-redpanda
force-pushed
the
start_offset
branch
from
39525c4 to
47d89f2
Compare
joe-redpanda
force-pushed
the
start_offset
branch
2 times, most recently
from
8df5d95 to
d24398c
Compare
|
/ci-repeat 1 |
joe-redpanda
force-pushed
the
start_offset
branch
from
d24398c to
544f866
Compare
|
/ci-repeat 1 |
joe-redpanda
force-pushed
the
start_offset
branch
from
544f866 to
f9f659f
Compare
|
/ci-repeat 1 |
joe-redpanda
force-pushed
the
start_offset
branch
from
f9f659f to
fea5344
Compare
|
/ci-repeat 1 |
joe-redpanda
force-pushed
the
start_offset
branch
from
fea5344 to
413b95d
Compare
Agreed. My thought was to leave this to bake for some time before performing the back-port. |
Short: Your start offset change will be visible right away, where all other updates will be visible whenever the queue gets burned down. Long: so for tp topic/1 you can have direct_consumer::get_start_offset("topic/1") -> 200 25.3 start offset was only getting used for metrics so this wasn't really an important issue. We decided the above was fine for the release but correctness should be fixed s.t. we can depend on the correct ordering of offsets. |
joe-redpanda
force-pushed
the
start_offset
branch
from
4803e96 to
4853013
Compare
|
dev rebase |
joe-redpanda
force-pushed
the
start_offset
branch
from
4853013 to
2619580
Compare
joe-redpanda
force-pushed
the
start_offset
branch
from
2619580 to
f62e336
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 9 changed files in this pull request and generated 6 comments.
Comments suppressed due to low confidence (2)
src/v/kafka/client/direct_consumer/fetcher.h:1
- Using -1 as an invalid/uninitialized marker for an epoch value may be problematic if -1 could ever be a valid epoch. Consider using
std::optional<kafka::leader_epoch>for fields that may not always be set.
/*
src/v/kafka/client/direct_consumer/fetcher.cc:1
- Corrected spelling of 'monatomic' to 'monotonic'.
/*
| // we'll keep attempting to pluck from the queue until the timeout is | ||
| // exhausted | ||
| while (ss::lowres_clock::now() < deadline) { | ||
| // either the remaining timeout or a small but reasonable minimum | ||
| // timeout |
There was a problem hiding this comment.
may I ask why this change is needed? It seems like this timeout should be honored within the data_queue API if nothing is available to be fetched.
There was a problem hiding this comment.
With filtering its possible that the next batch of fetch data from the queue is empty. The loop will continuously grab new fetches off the queue until it finds one that isn't entirely stale.
The deadline fiddling done here is so we don't join the cv's waiter queue with an already expired / imminently expiring timeout. If we're going through the work of calling into fetch_next, imo we should give the operation at least a task_quota to work with
| "offset filtering requires that unassigned subscriptions have " | ||
| "already been filtered out"); |
There was a problem hiding this comment.
nit: do we need to dump any debug state in the assert output
There was a problem hiding this comment.
I don't think so. This is a paranoid assert. I can feasibly only see this assert firing if a developer changes direct consumer, and the message is fairly clear in so far as I think it indicates
"you goofed up, this used to filter expired subscriptions and now it doesn't"
direct_consumer_fixture_test now only asserts on initial conditions in one test to reduce redundancy. The initial assert is changed to permit offset only updates, but restrict those updates to at max one per partition. direct_consumer_test is changed to no longer assert on empty fetches.
Adds utility functions to find subscriptions, returning the reference as an optional on reference wrapper. Updates usages of these to use the helpers for cleanliness
Add reasonable initializers to fetched_partition_data. This is not a required correctness change, instead this is meant to preempt a writer from forgetting to set a value when filling in the fields of fetched_partition_data. Adds are_offfsets_equal to source_partition_offsets which will check if tracked offsets are the same or different over time.
joe-redpanda
force-pushed
the
start_offset
branch
from
f62e336 to
7b3213d
Compare
This commit does four things 1. fetch data will now be added to the queue even if it has no batches 2. offsets will be updated at the point in time at which fetch_next is called 3. a new filter is applied to remove fetches which contain no new information (all offsets are the same) 4. fetch next will retry fetching from the data_queue if the filters have removed everything from the resultant fetch
The results of partitioning were incorrectly named in filter_stale_subscriptions. Fix the names and additionally pull the iterators from the subspan rather than doing wasteful (and dangerous) iterator math.
Clarifies why a vassert is firing in direct_consumer: Direct consumers filtering is order dependent to reduce the amount of code spent checking nullopt
Fetcher is remarkably error prone to work on. To mitigate this, this commit splits the logic for processing fetch responses into new delegated functions. 1. do_process_partition_response: a sync static method which is responsible for taking a given fetch and determining what should be done with it - retriable errors -> update metadata - out of bounds -> reset offsets - unknown error -> bubble to caller - data fetch -> return data - offset only fetch -> return offsets 2. process_partition_response: an async wrapper for do_process_partition_response which updates the fetcher local state and incorporates the results into the resultant fetch response
Adds fetcher unit tests to ensure the decision logic in do_process_partition_response is per expectations.
Adds clarity to the meaning and implication of consistent partitions in vasserts. Namely, the code is written to check whether a partition is consistent before operating on it, allowing us to skip most checks on iterators to the end of a collection and nullopts from helper getting methods. This was done to significantly cut down on invalid entry checks.
joe-redpanda
force-pushed
the
start_offset
branch
from
7b3213d to
140a8a3
Compare
A fetch is potentially stale until the point in time at which it is returned to the consumer in fetch_next.
This pr moves the update for offsets to only after the final subscription epoch filter has been applied to guarantee correctness of the provided offsets.
backport: this will get backported but we're going to bake the change before putting it through
Backports Required
Release Notes
Bug Fixes