kafka: add kafka_produce_batch_validation and rework timestamp validation#27529
Merged
rockwotj merged 4 commits intoredpanda-data:devfrom Sep 15, 2025
Merged
kafka: add kafka_produce_batch_validation and rework timestamp validation#27529rockwotj merged 4 commits intoredpanda-data:devfrom
kafka: add kafka_produce_batch_validation and rework timestamp validation#27529rockwotj merged 4 commits intoredpanda-data:devfrom
Conversation
WillemKauf
force-pushed
the
kba_set_max_ts
branch
from
c587808 to
3f44c51
Compare
Contributor
Maybe mention timequeries and time-based retention specifically? |
andrwng
reviewed
Sep 11, 2025
Contributor
andrwng
left a comment
andrwng
left a comment
There was a problem hiding this comment.
Nice, pretty much LGTM, just a couple thoughts
Comment on lines
+225
to
+226
| auto max_timestamp_missing = header.max_timestamp | ||
| == model::timestamp::missing() |
Contributor
There was a problem hiding this comment.
I imagine this is maybe a gray area, if the value is some arbitrary negative number, should we also treat it as missing?
Contributor
Author
There was a problem hiding this comment.
I imagine this is maybe a gray area, if the value is some arbitrary negative number, should we also treat it as missing?
Negative timestamps are supported by Kafka. https://cwiki.apache.org/confluence/display/KAFKA/KIP-228+Negative+record+timestamp+support
Contributor
There was a problem hiding this comment.
I thought that's just a proposal but indeed, just checked, kafka broker will happily accept negative timestamps.
Collaborator
CI test resultstest results on build#72099
test results on build#72124
test results on build#72154
test results on build#72220
test results on build#72221
test results on build#72226
test results on build#72227
test results on build#72262
|
WillemKauf
force-pushed
the
kba_set_max_ts
branch
from
3f44c51 to
a594333
Compare
WillemKauf
changed the title
kafka: set max_timestamp if == missing() in batch_adapterkafka: set or reject batches if max_timestamp is missing
Collaborator
Retry command for Build#72124please wait until all jobs are finished before running the slash command |
asimms41
reviewed
Sep 12, 2025
src/v/config/configuration.cc
Outdated
| , log_message_timestamp_reject_unset( | ||
| *this, | ||
| "log_message_timestamp_reject_unset", | ||
| "Controls whether batches produced to Redpanda with an unset " |
Contributor
There was a problem hiding this comment.
Maybe something like this:
Controls the handling of batches produced to Redpanda when the max_timestamp value is unset.
- When set to
false, each batch is accepted and themax_timestampis derived from the maximum timestamp of records in the batch. A WARN level log line is displayed to highlight potential additional costs when a batch is compressed and Redpanda must decompress it. - When set to
true, the batch is rejected and an ERROR level log line is displayed.
Contributor
Author
There was a problem hiding this comment.
Repushed with a new cluster property. Can I get a re-review please @asimms41?
asimms41
reviewed
Sep 12, 2025
Contributor
asimms41
left a comment
asimms41
left a comment
There was a problem hiding this comment.
Added suggestions for rewording.
nvartolomei
reviewed
Sep 12, 2025
Contributor
Author
|
@nvartolomei I have added a commit which rejects batches with |
WillemKauf
force-pushed
the
kba_set_max_ts
branch
2 times, most recently
from
9ff45ac to
9834600
Compare
WillemKauf
changed the title
kafka: set or reject batches if max_timestamp is missingkafka: set max_timestamp if missing
nvartolomei
reviewed
Sep 12, 2025
rockwotj
reviewed
Sep 12, 2025
Contributor
rockwotj
left a comment
rockwotj
left a comment
There was a problem hiding this comment.
Nice, LGTM, just one comment.
| }; | ||
|
|
||
| template<> | ||
| struct convert<model::kafka_batch_validation_mode> { |
Contributor
There was a problem hiding this comment.
so much boilerplate 😢
Thanks for doing this!
nvartolomei
reviewed
Sep 12, 2025
WillemKauf
force-pushed
the
kba_set_max_ts
branch
from
9834600 to
c00e5e2
Compare
rockwotj
requested review from
a team,
StephanDollberg and
piyushredpanda
and removed request for
a team and
piyushredpanda
Contributor
Author
|
/microbench |
Collaborator
|
Performance change detected in https://buildkite.com/redpanda/redpanda/builds/72274#01994e66-f605-4766-a1ff-09c3196414cc: See https://redpandadata.atlassian.net/wiki/x/LQAqLg for docs |
Member
|
Note there will be a lot of noise in these as we just changed instance type for microbenches. The kafka produce ones are probably real though. Best to confirm locally. |
Contributor
|
I think the change is mostly because the benchmark did not previously measure the work we were doing in the Kafka protocol layer to iterate over the batches. |
Contributor
|
Maybe e2e benchmarks is what we want? I guess we can see how those shake out in the nightly rights right @StephanDollberg? Is there a way to do the OMB benchmarking as a slash command? |
Contributor
Author
|
Before After |
Contributor
Author
|
Worth noting that this benchmark does not test the now compressed path ( |
Contributor
|
If we had a benchmark that measured the work done in kafka_batch_adapter and the produce path, I'd expect that delta to be much more reasonable. |
Contributor
|
@WillemKauf if you change the benchmark to use compressed batches with the default cluster config I'd expect the changes to go back to normal. Can you confirm? |
Member
Slash command OMB is not a yet thing. |
Contributor
Author
I assume you mean compressed batches with After with diff: ss::future<>
produce_partition_fixture::run_test(size_t data_size, measured_region region) {
BOOST_TEST_CHECKPOINT("HERE");
+ scoped_config cfg;
+ cfg.get("kafka_produce_batch_validation")
+ .set_value(model::kafka_batch_validation_mode::legacy);
model::topic_partition tp = model::topic_partition(
t, model::partition_id(0));
storage::record_batch_builder builder(
model::record_batch_type::raft_data, model::offset{0});
constexpr size_t num_records = 100;
for (size_t i = 0; i < num_records; ++i) {
builder.add_raw_kv(iobuf{}, rand_iobuf(data_size));
}
+ builder.set_compression(model::compression::lz4);
auto batch = std::move(builder).build(); |
Contributor
|
I'm a bit confused how those results are better than the baseline... Anyways, my recommendation is as follows:
|
Contributor
Author
|
Collaborator
|
/backport v25.2.x |
Collaborator
|
Failed to create a backport PR to v25.2.x branch. I tried: |
Member
Note the OMB benchmarks don't use compressed batches which to my understanding is the main regression path (besides from the normal one shown in the microbenchmark but we think that's only because of it now measuring more of what normally is on the produce path)? |
Contributor
I mean the java client also sets max timestamp so it also won't hit a perf impact unless strict validation is enabled. I mostly want to ensure that "well behaved clients" don't see a perf impact. |
Contributor
|
@WillemKauf, is it accurate to say this PR implements KIP-31 & KIP-32? |
Contributor
Author
No, it is not. |
Contributor
|
@WillemKauf, sorry, I should have posted a link: https://issues.apache.org/jira/browse/KAFKA-2511 |
Contributor
|
No it's not the same as that issue either. This is adding validation that isn't in a KIP but mirrors upstream Apache Kafka. |
This was referenced Sep 16, 2025
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
8 participants
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.
redpandacannot leavemax_timestampunset in record batches - several subsystems (timequeries, retention, archival) depend heavily on themax_timestampof batches being properly set.Set the
max_timestampmanually by iterating over the records in the record batch, taking the maximumtimestamp_deltaand adding it to the batch'sfirst_timestamp(which must be set by clients, as a hard rule). This may incur additional cost for compressed batches, and is gated by the level ofkafka_produce_batch_validation.kafka_produce_batch_validationis a new cluster property which shall be used in the future for more extensive batch validation in addition to the timestamp fixes here.Log lines are added to alert the client in case of a batch accepted without a maximum timestamp set for
kafka_produce_batch_validation == legacyor in the case of expensive decompression operation in the produce path forkafka_produce_batch_validation == relaxed, letting them know to update their client to set themax_timestamp.Opened for backports.
Backports Required
Release Notes
Features
kafka_produce_batch_validationwhich controls the level of validation performed on batches in theredpandaproduce path.