big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-17391. Adjust the checkpoint io buffer size to the chunk size (#6594). Contributed by lei w.

Browse Source 
Signed-off-by: He Xiaoqiao <hexiaoqiao@apache.org>
This commit is contained in:
Lei313 2024-03-12 18:36:43 +08:00 committed by GitHub
parent dbf08c872a
commit e211f6f83d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 15 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java
View File

@ -311,7 +311,7 @@ private static void uploadImage(URL url, Configuration conf,
ImageServlet.setVerificationHeadersForPut(connection, imageFile);
// Write the file to output stream.
writeFileToPutRequest(conf, connection, imageFile, canceler);
writeFileToPutRequest(conf, connection, imageFile, canceler, chunkSize);
int responseCode = connection.getResponseCode();
if (responseCode != HttpURLConnection.HTTP_OK) {
@ -330,7 +330,8 @@ private static void uploadImage(URL url, Configuration conf,
}
private static void writeFileToPutRequest(Configuration conf,
HttpURLConnection connection, File imageFile, Canceler canceler)
HttpURLConnection connection, File imageFile, Canceler canceler,
int bufferSize)
throws IOException {
connection.setRequestProperty(Util.CONTENT_TYPE, "application/octet-stream");
connection.setRequestProperty(Util.CONTENT_TRANSFER_ENCODING, "binary");
@ -338,7 +339,7 @@ private static void writeFileToPutRequest(Configuration conf,
FileInputStream input = new FileInputStream(imageFile);
try {
copyFileToStream(output, imageFile, input,
ImageServlet.getThrottler(conf), canceler);
ImageServlet.getThrottler(conf), canceler, bufferSize);
} finally {
IOUtils.closeStream(input);
IOUtils.closeStream(output);
@ -352,13 +353,14 @@ private static void writeFileToPutRequest(Configuration conf,
public static void copyFileToStream(OutputStream out, File localfile,
FileInputStream infile, DataTransferThrottler throttler)
throws IOException {
copyFileToStream(out, localfile, infile, throttler, null);
copyFileToStream(out, localfile, infile, throttler, null, -1);
}
private static void copyFileToStream(OutputStream out, File localfile,
FileInputStream infile, DataTransferThrottler throttler,
Canceler canceler) throws IOException {
byte buf[] = new byte[IO_FILE_BUFFER_SIZE];
Canceler canceler, int bufferSize) throws IOException {
int bufSize = bufferSize > 0 ? bufferSize : IO_FILE_BUFFER_SIZE;
byte[] buf = new byte[bufSize];
long total = 0;
int num = 1;
IOException ioe = null;
@ -369,13 +371,13 @@ private static void copyFileToStream(OutputStream out, File localfile,
.aboutToSendFile(localfile);
if (CheckpointFaultInjector.getInstance().
shouldSendShortFile(localfile)) {
// Test sending image shorter than localfile
long len = localfile.length();
buf = new byte[(int)Math.min(len/2, IO_FILE_BUFFER_SIZE)];
// This will read at most half of the image
// and the rest of the image will be sent over the wire
infile.read(buf);
shouldSendShortFile(localfile)) {
// Test sending image shorter than localfile
long len = localfile.length();
buf = new byte[(int) Math.min(len / 2, bufSize)];
// This will read at most half of the image
// and the rest of the image will be sent over the wire
infile.read(buf);
}
while (num > 0) {
if (canceler != null && canceler.isCancelled()) {