[CORE-10026] kafka: Support fetch v12#26926
[CORE-10026] kafka: Support fetch v12#26926BenPope wants to merge 3 commits intoredpanda-data:devfrom
Conversation
Signed-off-by: Ben Pope <ben@redpanda.com>
Signed-off-by: Ben Pope <ben@redpanda.com>
Populate the current leader id and epoch under certain error conditions. Signed-off-by: Ben Pope <ben@redpanda.com>
BenPope
requested review from
a team,
ballard26 and
michael-redpanda
and removed request for
a team
CI test resultstest results on build#69375
|
@redpanda-data/core-replication |
| template<std::ranges::input_range R> | ||
| auto enumerate(R&& range) { | ||
| return std::views::zip(std::views::iota(size_t{0}), std::forward<R>(range)); | ||
| } |
There was a problem hiding this comment.
Isn't this std::ranges::enumerate_view?
There was a problem hiding this comment.
Isn't this
std::ranges::enumerate_view?
Yes, good spot. (our libc++ doesn't have it), but I'll remove it as I'm not using it any more.
There was a problem hiding this comment.
enumerate is not supported in clang yet https://en.cppreference.com/w/cpp/compiler_support.html#cpp_lib_ranges_enumerate_202302L
| return needs_updated_leader(err) | ||
| .and_then([&](auto err) { | ||
| return get_leader_id_and_epoch(md_cache, ktp) | ||
| .transform([&](auto leader) { | ||
| return kafka::read_result(err, std::move(leader)); | ||
| }); | ||
| }) | ||
| .value_or(kafka::read_result(err)); |
There was a problem hiding this comment.
Interesting use of the optional utilities.
There was a problem hiding this comment.
Interesting use of the
optionalutilities.
Yeah, I'm not convinced!
| return needs_updated_leader(err) | |
| .and_then([&](auto err) { | |
| return get_leader_id_and_epoch(md_cache, ktp) | |
| .transform([&](auto leader) { | |
| return kafka::read_result(err, std::move(leader)); | |
| }); | |
| }) | |
| .value_or(kafka::read_result(err)); | |
| if (needs_updated_leader(err)) { | |
| if (auto leader = get_leader_id_and_epoch(md_cache, ktp); leader) { | |
| return kafka::read_result(err, std::move(leader)); | |
| } | |
| } | |
| return kafka::read_result(err); |
There was a problem hiding this comment.
i mean, it's your code. I am happy with the proposed changes. 👍
Without linearizable_barrier we may return stale data, i.e. an epoch (=term) that ceased to be correct at some point before we started serving the client request. |
|
I think it is fine not to issue the linearizable barrier. The term may only be stale, but it will not be stale if the replica serving a request is current leader. In other words there will be no inconsistency between the term and the leader_id |
I'd come to the same conclusion, but I appreciate the confirmation. |
|
Moving to closed, in favour of #26968 |
6 participants
Support Fetch v12
I wonder if the fetch should issue a
linearizable_barrierbefore returning the epoch, as is done forlist_offsets?Backports Required
Release Notes