Kafka batch reader #515

BenPope · 2021-01-27T00:10:14Z

This PR replaces #464

kafka::fetch_response::partition_response::record_set contains optional<record_batch[]>, which deserves a type.

This utility supports:

empty
size_bytes
last_offset - last offset of last batch
consume_batch - as a record_batch_adaptor
record_batch_reader::impl implementation
release() - release the underlying iobuf

Major differences since #464:

Renamed consumer_records to batch_reader
Renamed member consume_record_batch to consume_batch
Renamed member _record_set to _buf
Support in request_reader and response_writer
Maintain nullable semantics by inverting the optionality
Removed logging - record_batch adaptor already logs
Reordered the patchset as it's a much closer drop-in replacement

Major differences after force push:

Rebased
Introduce an exception hierarchy
Improved error handling with exceptions
Tests for failure modes

Checklist

Reference related issue
Update PendingReleaseNotes.md, if relevant

When referencing a related issue, remember to migrate duplicate stories from the
external tracker. This is not relevant for most users.

emaxerrno · 2021-01-27T03:06:48Z

src/v/kafka/protocol/batch_reader.cc

+struct record_batch_info {
+    model::offset base_offset{};
+    int32_t record_count{};
+    int32_t size_bytes{};


is this 0? why not set it to 0. what's the default initializer of an int.

i know sizet / unsigned are specified. are ints well specified in every compiler?

I think this is zero initialization via value initialization via default member initialization.

Pretty sure it's 0 for all integral types on all compilers that accept this as valid syntax.

Happy make it explicit for the benefit of doubt.

seems like a good candidate for using a constructor if the whole object is always initialized

emaxerrno · 2021-01-27T03:07:50Z

src/v/kafka/protocol/batch_reader.h

+/// model::record_batch on the wire
+///
+/// The array bound is not serialized, c.f. array<kafka::thing>
+class batch_reader final : public model::record_batch_reader::impl {


this looks really good.

dotnwat

looks good. i need to review again but i left a comment for discussion about the batch reader interface. it's not a lot of code so i'm not sure we need to spend a lot time shooting for the most generic interface, but it might be pretty easy to do because the pattern discussed is already available for copying from storage.

dotnwat · 2021-01-27T06:35:47Z

src/v/kafka/protocol/batch_reader.cc

+    return record_batch_info{
+      .base_offset = base_offset,
+      .record_count = record_count,
+      .size_bytes = size_bytes};


code here looks fine to me.

discussion: if this is all being done for speed, it might be even faster to index directly into the linearized bytes. and grab the three values. there is also a helpful little thing in the tree called PERF_TEST you could use to see if this code above saves any cycles compared to decoding the entire header.

src/v/kafka/protocol/batch_reader.cc

Lazin · 2021-01-27T07:35:51Z

src/v/kafka/protocol/tests/batch_reader_test.cc

+    auto crs = kafka::batch_reader(std::move(mem_res.data));
+
+    model::offset last_offset{};
+    while (!crs.empty()) {


Maybe check that the crc is not empty at the beginning?

It really should have 40 batches in it to start with. If it is empty, the latter test for last_offset should definitely fail.

dotnwat

looks great. minor feedback about a couple cosmetic things, and i think we need a test for the error cases of the record batch reader interface.

i also don't see the record batch reader interface being used other than in a test (am i missing something there?). if it isn't being used are you planning to use it later?

finally did you test some existing clients to make sure everything seems ok for other kafka clients?

dotnwat · 2021-01-27T22:24:21Z

src/v/kafka/protocol/batch_reader.cc

+struct record_batch_info {
+    model::offset base_offset{};
+    int32_t record_count{};
+    int32_t size_bytes{};


seems like a good candidate for using a constructor if the whole object is always initialized

dotnwat · 2021-01-27T22:26:31Z

src/v/kafka/protocol/batch_reader.cc

+        return size_bytes - model::packed_record_batch_header_size;
+    }
+    model::offset last_offset() {
+        return base_offset + model::offset{record_count - 1};


there is also the lastOffsetDelta field in the header that could be used here?

Sure, I had thought it was int64_t.

i was only suggesting that the header has a field specifically designed (i think!) to be added to base_offset to give the last offset. but importantly, i don't think that the two calculations are always the same after compaction runs. we should verify this though.

src/v/kafka/protocol/batch_reader.cc

src/v/kafka/protocol/tests/batch_reader_test.cc

src/v/kafka/protocol/batch_reader.cc

dotnwat · 2021-01-27T22:50:32Z

src/v/kafka/protocol/batch_reader.cc

+            if (likely(kba.v2_format && kba.valid_crc && kba.batch)) {
+                batches.push_back(std::move(*kba.batch));
+            } else {
+                _do_load_slice_failed = true;


clear batches here?

I was tempted, and then I can drop the _do_load_slice_failed member, WDYT?

I kinda like _do_load_slice_failed because it makes the termination condition clear but i don't have a strong opinion. Would it make sense to clear batches and immediately return an exceptional future? I'm not sure how you were planning on communicating an error to a consumer.

I thought about it as a better signalling mechanism, but a quick search through the codebase didn't reveal any other readers that do that, and I wasn't sure what was idiomatic.

I think log reader is the only record batch reader that has an opportunity to fail. Here is where it returns an exceptional future from do_load_slice

https://github.com/vectorizedio/redpanda/blob/dev/src/v/storage/log_reader.cc#L297

src/v/kafka/protocol/tests/batch_reader_test.cc

src/v/kafka/protocol/response_writer.h

src/v/kafka/server/tests/produce_consume_test.cc

BenPope · 2021-01-27T23:55:54Z

i also don't see the record batch reader interface being used other than in a test (am i missing something there?). if it isn't being used are you planning to use it later?

I think it will be useful here: https://github.com/vectorizedio/redpanda/pull/457/files#r563178111

Or if that link doesn't work: https://github.com/vectorizedio/redpanda/pull/457/files#diff-a918eb17ad60e95fda4bd6a66bcfc19ae0244a5ad9ead86732abc22c83e079afR207

And I have plans to use it for json serialization in fetch.

finally did you test some existing clients to make sure everything seems ok for other kafka clients?

No, but I probably should.

dotnwat · 2021-01-28T00:14:48Z

i also don't see the record batch reader interface being used other than in a test (am i missing something there?). if it isn't being used are you planning to use it later?

I think it will be useful here: https://github.com/vectorizedio/redpanda/pull/457/files#r563178111

Or if that link doesn't work: https://github.com/vectorizedio/redpanda/pull/457/files#diff-a918eb17ad60e95fda4bd6a66bcfc19ae0244a5ad9ead86732abc22c83e079afR207

Ah, nice. Makes sense

graphcareful

LGTM super nice impl

graphcareful · 2021-01-28T03:17:45Z

src/v/kafka/protocol/tests/batch_reader_test.cc

+        last_offset = kba.batch->last_offset();
+    }
+
+    BOOST_REQUIRE_EQUAL(last_offset, expected_last_offset);


Maybe it would also be useful to assert the number of bytes processed/read against the expected value

Signed-off-by: Ben Pope <[email protected]>

kafka::fetch_response::partition_response::record_set contains multiple record_batch, which deserves a type. This utility currently supports: * empty() * size_bytes() * release() - Release the underlying iobuf TODO: * last_offset - last offset of last batch * consume_batch - as a kafka_batch_adaptor * implement record_batch_reader::impl Signed-off-by: Ben Pope <[email protected]>

dotnwat

this patch looks great, nice work! it's ready to merge pending the question about if (unlikely(record_count - 1 != last_offset_delta))

dotnwat · 2021-01-28T23:26:38Z

src/v/kafka/protocol/batch_reader.cc

+        // they build the types bottoms up, not top down
+        + sizeof(base_offset) + sizeof(batch_length);
+
+    if (unlikely(record_count - 1 != last_offset_delta)) {


did you verify that this property always holds? i'm not sure that it does, but this might be relevant to the investigation IBM/sarama#1006

I was meant to remove that - it was initially there to check the implementation, and that if my understanding was correct. It makes sense that it may not hold for compacted topics. I'll remove it.

plz use --keep-base on the next rebase fixup :)

I don't usually move the base, I was somewhat forced into it (twice) :)

oh yeh i was just thinking that the next force push might be like a 3 line change

dotnwat · 2021-01-28T23:29:35Z

src/v/kafka/protocol/tests/batch_reader_test.cc

+                                     sizeof(int8_t);         // magic
+static constexpr size_t lod_offset = crc_offset +            //
+                                     sizeof(int32_t) +       // crc
+                                     sizeof(int16_t);        // attr


dotnwat · 2021-01-28T23:36:18Z

src/v/kafka/protocol/batch_reader.cc

+                    "Invalid kafka record parsing: {}",
+                    !kba.v2_format   ? "not v2_format"
+                    : !kba.valid_crc ? "invalid crc"
+                                     : "empty batch")));


this looks great

I think I got the correct order - it makes sense that magic is checked before crc, right?

does the order affect anything other than what message gets printed? seems like you could also do like "crc {} magic {} empty {}", crc, magic, empty or something along those lines?

It's just the printed message, but I like your version more.

I considered adding an enum (or flags) for failure mode, and an exception type to hold it - which I could check it in the test. But I thought the tests would be the only place it would be used.

ok, so valid_crc isn't set if magic != 2

runtime error: load of value 27, which is not a valid value for type 'const bool'

I'll leave it as it was.

dotnwat · 2021-01-28T23:36:53Z

src/v/kafka/protocol/tests/batch_reader_test.cc

+      model::consume_reader_to_memory(std::move(rdr), model::no_timeout).get(),
+      kafka::exception,
+      [](const kafka::exception& e) {
+          return e.error == kafka::error_code::corrupt_message;


these failure test cases are really nice

Obtain the last offset of the last batch Note: This relies on kafka_batch_adaptor which currently lives in v_kafka Signed-off-by: Ben Pope <[email protected]>

Consume a single record_batch as a kafka_batch_adaptor Signed-off-by: Ben Pope <[email protected]>

Signed-off-by: Ben Pope <[email protected]>

Replaces fetch_response::partition_response::record_set Signed-off-by: Ben Pope <[email protected]>

Now that obtaining the last offset isn't destructive, test that all produced records are consumed. Signed-off-by: Ben Pope <[email protected]>

Drive-by fix: Remove some dead code. Signed-off-by: Ben Pope <[email protected]>

BenPope requested review from dotnwat, graphcareful and emaxerrno as code owners January 27, 2021 00:10

BenPope force-pushed the kafka-batch-reader branch 2 times, most recently from f49cf52 to 5be75ae Compare January 27, 2021 00:45

emaxerrno reviewed Jan 27, 2021

View reviewed changes

dotnwat reviewed Jan 27, 2021

View reviewed changes

Lazin reviewed Jan 27, 2021

View reviewed changes

BenPope mentioned this pull request Jan 27, 2021

k/requests: Introduce consumer_records #464

Closed

2 tasks

dotnwat reviewed Jan 27, 2021

View reviewed changes

graphcareful approved these changes Jan 28, 2021

View reviewed changes

BenPope added 2 commits January 28, 2021 21:19

k/protocol: Introduce an exception hierarchy

0d4bfde

Signed-off-by: Ben Pope <[email protected]>

BenPope force-pushed the kafka-batch-reader branch from 5be75ae to 916a6a4 Compare January 28, 2021 21:39

BenPope requested review from dotnwat and emaxerrno January 28, 2021 22:41

dotnwat reviewed Jan 28, 2021

View reviewed changes

BenPope added 7 commits January 29, 2021 00:21

k/protocol: Introduce batch_reader::last_offset

98ec4d6

Obtain the last offset of the last batch Note: This relies on kafka_batch_adaptor which currently lives in v_kafka Signed-off-by: Ben Pope <[email protected]>

k/protocol: Introduce batch_reader::consume_batch

ea80c28

Consume a single record_batch as a kafka_batch_adaptor Signed-off-by: Ben Pope <[email protected]>

k/protocol: Make batch_reader implement record_batch_reader::impl

e8ae554

Signed-off-by: Ben Pope <[email protected]>

k/protocol: Support batch_reader in request_reader, response_writer

af2b2f9

Signed-off-by: Ben Pope <[email protected]>

k/protocol: Switch fetch_response to batch_reader

dd4159c

Replaces fetch_response::partition_response::record_set Signed-off-by: Ben Pope <[email protected]>

k/server/tests: Improve produce_consume_test

e60ed26

Now that obtaining the last offset isn't destructive, test that all produced records are consumed. Signed-off-by: Ben Pope <[email protected]>

pp/json: Test fetch: Use batch_reader for make_record_ret

87771a3

Drive-by fix: Remove some dead code. Signed-off-by: Ben Pope <[email protected]>

BenPope force-pushed the kafka-batch-reader branch from 916a6a4 to 87771a3 Compare January 29, 2021 00:59

BenPope requested a review from dotnwat January 29, 2021 01:02

dotnwat approved these changes Jan 29, 2021

View reviewed changes

BenPope mentioned this pull request Jan 29, 2021

Pandaproxy consumer groups fetch #456

Closed

2 tasks

dotnwat merged commit d7e1dcb into redpanda-data:dev Jan 29, 2021

Kafka batch reader #515

Kafka batch reader #515

Conversation

BenPope commented Jan 27, 2021 • edited Loading

Checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BenPope Jan 27, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat Jan 27, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BenPope commented Jan 27, 2021 • edited Loading

dotnwat commented Jan 28, 2021

graphcareful left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BenPope Jan 29, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BenPope commented Jan 27, 2021 •

edited

Loading

BenPope Jan 27, 2021 •

edited

Loading

dotnwat Jan 27, 2021 •

edited

Loading

BenPope commented Jan 27, 2021 •

edited

Loading

BenPope Jan 29, 2021 •

edited

Loading