Non-polling fetch implementation

[ballard26]

This PR implements a new fetch_plan_executor that doesn't repeatedly poll every partition in the fetch for new data. Instead it relies on registering callbacks with raft:consensus::visible_offset_monitor to know when a partition has new data and therefore would be worth querying once again.

The gist of how this is implemented is as follows;

• A fetch "coordinator"(see kafka::nonpolling_fetch_plan_executor::execute_plan) is created on the shard that received the fetch request. This coordinator is responsible for creating fetch workers and determining when a fetch request is completed.
• A fetch "worker"(see kafka::nonpolling_fetch_plan_executor::shard_fetch_worker) is created on every shard that has a partition from the request. It's responsible for querying partitions for data. And if no partitions have data it'll register with the raft:consensus::visible_offset_monitor for those partitions and wait until it increases for one or more of them. The worker only returns on errors, aborts from the worker, or when it has queried enough data to meet or exceed the lower limit the coordinator specified.

Backports Required

• none - not a bug fix
• none - this is a backport
• none - issue does not exist in previous branches
• none - papercut/not impactful enough to backport
• v23.3.x
• v23.2.x
• v23.1.x
• v22.3.x

Release Notes

Improvements

• Introduces a new non-polling fetch execution strategy that decreases CPU utilization of fetch requests and fetch request latency.
• Adds a new cluster configuration property fetch_read_strategy. This property determines which fetch execution strategy Redpanda will use to fulfill a fetch request. The newly introduced non_polling execution strategy is the default for this property with the polling strategy being included to make backporting possible.

[ballard26]

/dt

[travisdowns]

Looks good to me, but at least some of the test failures look legit with BadLogLines related to the switch to ERROR logging.

As you add cases to handle those, can you also add comments as to why we expect certain error types in that function?

[travisdowns]

Awesome, let the new non-polling fetch era begin!

[vbotbuildovich]

/backport v23.3.x

[rockwotj]

FWIW I feel like this is a little error prone. Mostly in this layer we're using kafka offsets (that have gone through translation), but here we explicitly are using raft offsets.

We should probably just prioritize using kafka::offset properly so this footgun is more explicit, but some documentation on this might help.

[rockwotj]

Otherwise a quick look at this PR looks good. Nice work!