verifier: tolerate data loss

[nvartolomei]

• Introduce a -continuous flag which instructs kgo-verifier to continue polling for new data rather than exiting or looping.
• Introduce a -tolerate-data-loss flag which facilitates testing write caching feature where data loss is tolerable in specific circumstances.

[andrwng]

LGTM! Just nits for me.

Also, just want to clarify the behavior when we see data loss:

• validator is running with -continuous and -tolerate-data-loss
• validator consumes up to 100, and is continuing to poll for 101+
• data is "rolled back" to 90
• PollFetches is doing some a periodic poll of highest offset, and under the hood will detect it moving backwards...? Is that right?
• validator detects this, adjusts the ranges, and continues with respective workloads

[nvartolomei]

LGTM! Just nits for me.

Also, just want to clarify the behavior when we see data loss:

• validator is running with -continuous and -tolerate-data-loss
• validator consumes up to 100, and is continuing to poll for 101+
• data is "rolled back" to 90
• PollFetches is doing some a periodic poll of highest offset, and under the hood will detect it moving backwards...? Is that right?
• validator detects this, adjusts the ranges, and continues with respective workloads

Correct. When offsets are "rolled back" we necessarily increment the raft term / leader epoch. When Redpanda receives a fetch request with an "outdated" epoch, it responds with fenced_leader_epoch (https://github.com/redpanda-data/redpanda/blob/c2e22debf59dbc4ad91dfa8d1a9520e352b382fe/src/v/kafka/server/handlers/details/leader_epoch.h#L34). Note: This happens pre-write caching too.

With this we (franz-go) fetches the last offset for that epoch using a offset_for_leader_epoch request (https://github.com/redpanda-data/redpanda/blob/40f4ad32381fec89043281805f2877bf4d84b3fc/src/v/kafka/server/handlers/offset_for_leader_epoch.cc#L122) / kerr.DataLoss on kgo side which tells us where to rollback