Skip to content

Commit

Permalink
[MINOR] improvement(server): LocalFileNioWriter support writeIndex api (
Browse files Browse the repository at this point in the history
#2320)

### What changes were proposed in this pull request?

Support WriteIndex api for LocalFileNioWriter.

### Why are the changes needed?

- Finish the suggestion left in #2223 (comment) 
- Introduce the `LocalFileNullDeviceWriter` in the follow up PR.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

No need.
  • Loading branch information
maobaolong authored Jan 6, 2025
1 parent a669c2b commit 995de6d
Showing 1 changed file with 13 additions and 3 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.uniffle.storage.handler.impl;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
Expand All @@ -30,6 +32,7 @@

public class LocalFileNioWriter implements FileWriter {

private DataOutputStream dataOutputStream;
private FileOutputStream fileOutputStream;
private long nextOffset;

Expand All @@ -40,6 +43,8 @@ public LocalFileNioWriter(File file) throws IOException {

public LocalFileNioWriter(File file, int bufferSize) throws IOException {
fileOutputStream = new FileOutputStream(file, true);
// init fsDataOutputStream
dataOutputStream = new DataOutputStream(new BufferedOutputStream(fileOutputStream, bufferSize));
nextOffset = file.length();
}

Expand All @@ -61,7 +66,12 @@ public void writeData(ByteBuf buf) throws IOException {

@Override
public void writeIndex(FileBasedShuffleSegment segment) throws IOException {
throw new UnsupportedOperationException("LocalFileNioWriter does not support index");
dataOutputStream.writeLong(segment.getOffset());
dataOutputStream.writeInt(segment.getLength());
dataOutputStream.writeInt(segment.getUncompressLength());
dataOutputStream.writeLong(segment.getCrc());
dataOutputStream.writeLong(segment.getBlockId());
dataOutputStream.writeLong(segment.getTaskAttemptId());
}

@Override
Expand All @@ -71,8 +81,8 @@ public long nextOffset() {

@Override
public synchronized void close() throws IOException {
if (fileOutputStream != null) {
fileOutputStream.close();
if (dataOutputStream != null) {
dataOutputStream.close();
}
}
}

0 comments on commit 995de6d

Please sign in to comment.