KAFKA-10694: Implement zero copy for FetchSnapshot by dengziming · Pull Request #9819 · apache/kafka

dengziming · 2021-01-03T09:31:29Z

More detailed description of your change

RawSnapshotWriter and RawSnapshotReader interact with BaseRecords instead of ByteBuffer
Use fileRecords.slice when read a snapshot
No big change of writing a snapshot

Summary of testing strategy (including rationale)

Change code in unit test of RawSnapshotWriter and RawSnapshotReader
Add unit test in SendBuilderTest and FileRawSnapshotTest
Add unit test for FileRegion and MemoryRegion

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

hachikuji · 2021-01-06T00:24:57Z

@dengziming Thanks for the patch. I have one high level comment. The approach we've taken here is to implement a new zero-copy "region" type. One alternative is to use the existing "records" type since it already supports "zero-copy" and the snapshot data is expected to be records data anyway. I'd like to understand the drawbacks of this approach a bit better. I think having two zero-copy types is a bit undesirable. An alternative is that we can let the "region" type subsume "records," but I suspect that will be tricky because of the down-conversion logic.

@jsancio also suggested that we might be able to represent the "region" type using type=bytes zeroCopy=true. I think that would require us to introduce an interface of some kind so that we can represent both a ByteBuffer and FileChannel. I'm not sure what that would look like, but it's also worth considering.

Anyway, just want to make sure we have covered all of the options before getting too far with this.

hachikuji · 2021-01-15T18:49:26Z

@dengziming @jsancio I'll give my take what I think we should do here, but let's try to agree before we go any further. Basically I am not really convinced that we need a new type, at least not in the near term. The snapshot data will be record data, so I cannot see a strong argument not to use the existing "records" type which already handles zero-copy.

The one difference between FetchSnapshot and Fetch is that the latter is aligned on offsets while the former can indicate an arbitrary position in a file. I assume that is the main reason we have diverged here. However, I do not think this presents a major hurdle. On the server side, I think we have everything we need. FileRawSnapshotReader already has a FileRecords object and we can return a slice at an arbitrary position using slice(int position, int size). On the client side, when we parse a snapshot response, Readable will give us back a reference to a MemoryRecords which will not necessarily be aligned. However, MemoryRecords gives us direct access to the underlying buffer, so the misalignment does not cause any problems.

Longer term, I think we can make an effort to replace the "records" type and just rely on type=bytes, zeroCopy=true. For this to work, we need a type which generalizes both a ByteBuffer and a FileChannel slice. For example, we could have something like this:

interface ZeroCopyBuffer {
  ByteBuffer read(int length);
  Send toSend();
}

When parsing a request or response received over the network, it will be in memory anyway, so we need some hook to get access to the ByteBuffer. On the other hand, when transmitting a request or response, we let the implementation build the Send itself. This gives us a way to work with RecordsSend. There's probably more to it than that, but that's probably enough to get started. Then we would change the generated code to rely on the new type and let SendBuilder invoke ZeroCopyBuffer.toSend as appropriate.

Thoughts?

jsancio · 2021-01-15T20:33:52Z

Replying to #9819 (comment)

Thanks @hachikuji for thinking about this and for giving a detail reply. For the long term solution, I am trying to understand how we transition from type: "records" to type: "bytes", zeroCopy="true".

Outside of what you have enumerate, how would changing the type of a field work after the version bump? Would the generated message class need to support access to this field as both BaseRecords and ZeroCopyBuffer? Or would the generated class would just have ZeroCopyBuffer? In other words, what is the expectation of 3rd party clients with regards to the generated classes?

hachikuji · 2021-01-15T22:09:41Z

@jsancio It's a good question. Up to now, we have not enforced strict compatibility for the JSON schema representation. As long as the on-the-wire bytes are the same, we have allowed some flexibility for the generator. It was only in the past couple months that we finally got all of the APIs converted, so the flexibility was necessary. Personally I'd like to retain this loose compatibility going forward, but I guess it will depend on what users want. If there is a strong demand to keep the schema definitions compatible, then we will need to come up with some way to evolve them. However, for now, I don't think we should consider JSON type compatibility to be a blocker.

ijuma · 2021-01-15T22:11:23Z

The json schema is indeed not public api today. A KIP will be required before we make it so.

jsancio · 2021-01-16T02:15:42Z

The proposal in #9819 (comment) sounds good to me. I'll update the associated Jira soon.

dengziming · 2021-01-16T04:03:20Z

comment

I have tried locally to replace the FileRegion with FileRecords, and it worked fine, which means that "aligned on offsets" is just a soft requirement of FetchRequest, the MemoryRecords from FetchResponse is not necessarily be aligned, so it's viable to use FileRecords and MemoryRecords to implement zero-copy, the only additional work is to support BaseRecords.writeTo(FileChannel)

dengziming · 2021-01-19T06:20:04Z

@hachikuji @jsancio @ijuma ,Hello, This pr is ready for a second review, Thank you.

jsancio

Thanks for the change. Partial review. I wanted to give you some high-level comments early.

hachikuji · 2021-01-22T19:41:16Z

@dengziming Thanks for the updates. I think we're on the right track. I thought of a few potential simplifications. Take a look here: hachikuji@01cc3aa.

dengziming · 2021-01-24T04:32:28Z

@hachikuji ,Thank you for your suggestions, I found DefaultRecordsSend similar to UnalignedRecordsSend so I added a generic type toDefaultRecordsSend, and also added a parent interface TransferableRecords to Records and UnalignedRecords, please take a look.

hachikuji

Thanks, this is looking good. Left a few more comment, but I think we're almost ready to wrap this up.

hachikuji · 2021-01-25T06:52:15Z

+    }
+
+    @Override
+    public long writeTo(TransferableChannel destChannel, long previouslyWritten, int remaining) throws IOException {


Do we have test coverage for this?

Add UnalignedFileRecordsTest

Hmm. The semantic between the in-memory implementation vs the file channel implementation is slightly different. The in-memory version throws an exception if there are not enough bytes to send length. While the file channel version just sends less bytes if there is not enough to send remaining.

I suspect that we want to throw in both cases else the calling code won't know why writeTo sent less bytes. It could loop forever if it is attempting to send up to remaining bytes.

@jsancio , I think you are right, I also find that the implementation of MemoryRecords and FileRecords is also different, I will spend some time checking this.

hachikuji

Thanks, should be the last round of reviews (at least from me).

hachikuji · 2021-01-25T17:41:42Z

+    @Test
+    public void testWriteTo() throws IOException {
+
+        org.apache.kafka.common.requests.ByteBufferChannel channel = new org.apache.kafka.common.requests.ByteBufferChannel(fileRecords.sizeInBytes());


nit: why not import ByteBufferChannel?

ImportControl doesn't allow us to import ByteBufferChannel here, this is similar to NonOverflowingByteBufferChannel in your pr #4574

jsancio

@dengziming thanks for the changes. @hachikuji thanks for the reviews.

jsancio · 2021-01-25T23:49:35Z

+    }
+
+    @Override
+    public long writeTo(TransferableChannel destChannel, long previouslyWritten, int remaining) throws IOException {


Hmm. The semantic between the in-memory implementation vs the file channel implementation is slightly different. The in-memory version throws an exception if there are not enough bytes to send length. While the file channel version just sends less bytes if there is not enough to send remaining.

I suspect that we want to throw in both cases else the calling code won't know why writeTo sent less bytes. It could loop forever if it is attempting to send up to remaining bytes.

jsancio · 2021-01-26T00:02:42Z

+
+        ByteBuffer dup = sourceBuffer.duplicate();
+        dup.position(position);
+        dup.limit(position + length);


Isn't this position() and limit() modification the same as ByteBuffer.slice?

Having said that I find it strange that we want to set the absolute position and limit. Shouldn't this be set relative to the current position? For example think about sourceBuffer.position() > position.

There are many occurrences of position+limit in Kafka project, I think this is because the developers of Kafka like to handle the ByteBuffer by ourselves.

jsancio · 2021-01-26T00:05:13Z

+    public static long tryWriteTo(TransferableChannel destChannel,
+                                  int position,
+                                  int length,
+                                  ByteBuffer sourceBuffer) throws IOException {


How about removing the IO from this method with a signature like:

public static ByteBuffer relativeSlice(ByteBuffer buffer, int position, int length);

The caller of this method can then long written = destChannel.write(Utils.relativeSlice(buffer, position, length)).

hachikuji

LGTM. Thanks for the great contribution!

@jsancio If you have any remaining concerns, let's address them in follow-up patches.

dengziming · 2021-01-27T02:37:58Z

@hachikuji @jsancio Thank you for your great patience!
@jsancio your concerns seems similar to #9970 (comment) , we can discuss it there.

ijuma reviewed Jan 6, 2021

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/record/FileRegion.java Outdated

dengziming changed the title ~~KAFKA-10694: Implement zero copy for FetchSnapshot~~ [WIP] KAFKA-10694: Implement zero copy for FetchSnapshot Jan 9, 2021

dengziming force-pushed the KAFKA-10694 branch from ee15035 to 74ba026 Compare January 17, 2021 14:51

dengziming changed the title ~~[WIP] KAFKA-10694: Implement zero copy for FetchSnapshot~~ KAFKA-10694: Implement zero copy for FetchSnapshot Jan 18, 2021

jsancio reviewed Jan 19, 2021

View reviewed changes

Comment thread raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java Outdated

Comment thread raft/src/main/java/org/apache/kafka/snapshot/FileRawSnapshotWriter.java Outdated

Comment thread clients/src/main/java/org/apache/kafka/common/record/BaseRegion.java Outdated

implement zero-copy using BaseRecords

3f604b4

hachikuji reviewed Jan 21, 2021

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/utils/Utils.java Outdated

Comment thread raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java Outdated

Comment thread raft/src/main/java/org/apache/kafka/snapshot/FileRawSnapshotWriter.java Outdated

resolve comments

7f58dd4

dengziming force-pushed the KAFKA-10694 branch from 74ba026 to 7f58dd4 Compare January 21, 2021 09:54

hachikuji reviewed Jan 21, 2021

View reviewed changes

add UnalignedRecords

0ccd53d

Add unaligned records

d2025a2

dengziming added 3 commits January 24, 2021 12:46

minor

2e76f4b

Merge branch 'trunk' into KAFKA-10694

6b28107

fix unused code

79c1209

hachikuji reviewed Jan 25, 2021

View reviewed changes

resolve comments

2ead903

dengziming requested a review from hachikuji January 25, 2021 11:49

hachikuji reviewed Jan 25, 2021

View reviewed changes

jsancio reviewed Jan 26, 2021

View reviewed changes

dengziming added 2 commits January 26, 2021 11:43

resolve jason's comments

a3b29e7

resolve qa problems

d75d65b

hachikuji approved these changes Jan 26, 2021

View reviewed changes

hachikuji merged commit a26db2a into apache:trunk Jan 26, 2021

dengziming deleted the KAFKA-10694 branch October 8, 2022 11:51

Uh oh!

Conversation

dengziming commented Jan 3, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji commented Jan 6, 2021

Uh oh!

Uh oh!

hachikuji commented Jan 15, 2021

Uh oh!

jsancio commented Jan 15, 2021

Uh oh!

hachikuji commented Jan 15, 2021

Uh oh!

ijuma commented Jan 15, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jsancio commented Jan 16, 2021

Uh oh!

dengziming commented Jan 16, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dengziming commented Jan 19, 2021

Uh oh!

jsancio left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hachikuji commented Jan 22, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dengziming commented Jan 24, 2021

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jsancio left a comment

dengziming commented Jan 3, 2021 •

edited

Loading

ijuma commented Jan 15, 2021 •

edited

Loading

dengziming commented Jan 16, 2021 •

edited

Loading

hachikuji commented Jan 22, 2021 •

edited

Loading