FMX commented on code in PR #3349:
URL: https://github.com/apache/celeborn/pull/3349#discussion_r2197123896
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java:
##########
@@ -296,9 +319,22 @@ protected void readHeaderOrIndexBuffer(FileChannel
channel, ByteBuffer buffer, i
buffer.flip();
}
- protected void readBufferIntoReadBuffer(FileChannel channel, ByteBuf buf,
int length)
+ protected void readHeaderOrIndexBuffer(
+ FSDataInputStream inputStream, ByteBuffer buffer, long fileSize, int
length, String filePath)
throws IOException {
- Utils.checkFileIntegrity(channel, length);
+ Utils.checkFileIntegrity(fileSize - inputStream.getPos(), length,
filePath);
+ buffer.clear();
+ buffer.limit(length);
+ while (buffer.hasRemaining()) {
+ inputStream.read(buffer);
+ }
+ buffer.flip();
+ }
+
+ protected void readBufferIntoReadBuffer(
Review Comment:
These two methods have identical logic, maybe we can merge them.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java:
##########
@@ -307,19 +343,55 @@ protected void readBufferIntoReadBuffer(FileChannel
channel, ByteBuf buf, int le
buf.writeBytes(tmpBuffer);
}
- protected int readBuffer(
- String filename, FileChannel channel, ByteBuffer header, ByteBuf buffer,
int headerSize)
+ protected void readBufferIntoReadBuffer(
+ FSDataInputStream inputStream, ByteBuf buf, long fileSize, int length,
String filePath)
throws IOException {
- readHeaderOrIndexBuffer(channel, header, headerSize);
+ Utils.checkFileIntegrity(fileSize - inputStream.getPos(), length,
filePath);
+ ByteBuffer tmpBuffer = ByteBuffer.allocate(length);
+ while (tmpBuffer.hasRemaining()) {
+ inputStream.read(tmpBuffer);
+ }
+ tmpBuffer.flip();
+ buf.writeBytes(tmpBuffer);
+ }
+
+ protected int readBuffer(FSDataInputStream inputStream, ByteBuf buffer)
throws IOException {
Review Comment:
The following methods have the same problem.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
