hachikuji commented on a change in pull request #9819:
URL: https://github.com/apache/kafka/pull/9819#discussion_r563488709
##########
File path: clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
##########
@@ -134,6 +134,16 @@ public void readInto(ByteBuffer buffer, int position)
throws IOException {
* @return A sliced wrapper on this message set limited based on the given
position and size
*/
public FileRecords slice(int position, int size) throws IOException {
+ SliceRange range = sliceRange(position, size);
+ return new FileRecords(file, channel, range.start, range.end, true);
+ }
+
+ public UnalignedFileRecords sliceUnaligned(int position, int size) {
Review comment:
Can we add method documentation? I think the key thing to note is that
this method is reserved for cases where the position is not necessarily on an
offset boundary. It is worth mentioning the `FetchSnapshot` use case explicitly.
##########
File path: clients/src/main/java/org/apache/kafka/common/record/SliceRange.java
##########
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.record;
+
+
+public class SliceRange {
Review comment:
If this is only used in `FileRecords`, perhaps we can move it there and
make it `private static`.
##########
File path:
clients/src/main/java/org/apache/kafka/common/protocol/SendBuilder.java
##########
@@ -137,6 +138,9 @@ public void writeRecords(BaseRecords records) {
if (records instanceof MemoryRecords) {
flushPendingBuffer();
addBuffer(((MemoryRecords) records).buffer());
+ } else if (records instanceof UnalignedMemoryRecords) {
Review comment:
Are there any tests we can update to cover this?
##########
File path: clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
##########
@@ -144,11 +154,12 @@ public FileRecords slice(int position, int size) throws
IOException {
if (size < 0)
throw new IllegalArgumentException("Invalid size: " + size + " in
read from " + this);
- int end = this.start + position + size;
+ int sliceStart = this.start + position;
+ int sliceEnd = sliceStart + size;
// handle integer overflow or if end is beyond the end of the file
- if (end < 0 || end > start + currentSizeInBytes)
- end = start + currentSizeInBytes;
- return new FileRecords(file, channel, this.start + position, end,
true);
+ if (sliceEnd < 0 || sliceEnd > start + currentSizeInBytes)
+ sliceEnd = start + currentSizeInBytes;
+ return new SliceRange(sliceStart, sliceEnd);
Review comment:
I was thinking how we can simplify this a little bit and avoid the
annoying object. I think the main logic here is validating the input and
computing the size of the available bytes to read.
```java
private int availableBytes(int position, int maxLength) {
...
int limit = this.start + position + size;
if (limit < 0 || limit > start + currentSizeInBytes)
limit = start + currentSizeInBytes;
return limit - position;
}
```
Then the caller can do something like this:
```
int availableBytes = availableBytes(position, size);
return new UnalignedFileRecords(channel, this.start + position,
availableBytes);
```
What do you think? Any better?
##########
File path: clients/src/main/java/org/apache/kafka/common/utils/Utils.java
##########
@@ -1098,6 +1099,23 @@ public static void writeFully(FileChannel channel,
ByteBuffer sourceBuffer) thro
channel.write(sourceBuffer);
}
+ public static long tryWriteTo(TransferableChannel destChannel,
Review comment:
nit: add documentation?
##########
File path:
clients/src/main/java/org/apache/kafka/common/record/TransferableRecords.java
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.record;
+
+import org.apache.kafka.common.network.TransferableChannel;
+
+import java.io.IOException;
+
+/**
+ * Represents a record set which can be transfer to a channel
Review comment:
nit:
> Represents a record set which can be transferred to a channel.
##########
File path:
clients/src/main/java/org/apache/kafka/common/record/UnalignedFileRecords.java
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.record;
+
+import org.apache.kafka.common.network.TransferableChannel;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+
+/**
+ * Represents a file record set which is not necessarily offset-aligned
+ */
+public class UnalignedFileRecords implements UnalignedRecords {
+
+ private final FileChannel channel;
+ private final long position;
+ private final int size;
+
+ public UnalignedFileRecords(FileChannel channel, long position, int size) {
+ this.channel = channel;
+ this.position = position;
+ this.size = size;
+ }
+
+ @Override
+ public int sizeInBytes() {
+ return size;
+ }
+
+ @Override
+ public long writeTo(TransferableChannel destChannel, long
previouslyWritten, int remaining) throws IOException {
Review comment:
Do we have test coverage for this?
##########
File path:
raft/src/main/java/org/apache/kafka/snapshot/FileRawSnapshotWriter.java
##########
@@ -53,13 +54,13 @@ public long sizeInBytes() throws IOException {
}
@Override
- public void append(ByteBuffer buffer) throws IOException {
+ public void append(UnalignedMemoryRecords records) throws IOException {
Review comment:
It might be helpful to have an `append(MemoryRecords)`. That saves an
annoying conversion when we are generating the snapshot instead of copying it
from the leader.
##########
File path: clients/src/main/java/org/apache/kafka/common/utils/Utils.java
##########
@@ -1098,6 +1099,23 @@ public static void writeFully(FileChannel channel,
ByteBuffer sourceBuffer) thro
channel.write(sourceBuffer);
}
+ public static long tryWriteTo(TransferableChannel destChannel,
+ long position,
+ int length,
+ ByteBuffer sourceBuffer) throws IOException {
+ if (position > Integer.MAX_VALUE)
Review comment:
Is this a restriction of this method itself or are we just carrying over
the limitation from `FileRecords`?
##########
File path: clients/src/main/resources/common/message/FetchSnapshotResponse.json
##########
@@ -51,8 +51,8 @@
"about": "The total size of the snapshot." },
{ "name": "Position", "type": "int64", "versions": "0+",
"about": "The starting byte position within the snapshot included in
the Bytes field." },
- { "name": "Bytes", "type": "bytes", "versions": "0+", "zeroCopy": true,
- "about": "Snapshot data." }
+ { "name": "Bytes", "type": "records", "versions": "0+",
Review comment:
I had a previous comment about changing the name here from "Bytes" to
"UnalignedRecords." What do you think about it?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
