From 4abb2ba58c3ad059c00a46b7bc567f0fb192608c Mon Sep 17 00:00:00 2001 From: Ashutosh Gupta Date: Thu, 23 Jun 2022 17:28:32 +0100 Subject: [PATCH] YARN-10320.Replace FSDataInputStream#read with readFully in Log Aggregation (#4486) * YARN-10320.Replace FSDataInputStream#read with readFully in Log Aggregation Co-authored-by: Ashutosh Gupta --- .../LogAggregationIndexedFileController.java | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/filecontroller/ifile/LogAggregationIndexedFileController.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/filecontroller/ifile/LogAggregationIndexedFileController.java index 6cd1f53d4a..d4431d56b3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/filecontroller/ifile/LogAggregationIndexedFileController.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/filecontroller/ifile/LogAggregationIndexedFileController.java @@ -282,7 +282,8 @@ private Pair initializeWriterInRolling( checksumFileInputStream = fc.open(remoteLogCheckSumFile); int nameLength = checksumFileInputStream.readInt(); byte[] b = new byte[nameLength]; - int actualLength = checksumFileInputStream.read(b); + checksumFileInputStream.readFully(b); + int actualLength = b.length; if (actualLength == nameLength) { String recoveredLogFile = new String( b, Charset.forName("UTF-8")); @@ -765,7 +766,8 @@ public Map parseCheckSumFiles( checksumFileInputStream = fc.open(file.getPath()); int nameLength = checksumFileInputStream.readInt(); byte[] b = new byte[nameLength]; - int actualLength = checksumFileInputStream.read(b); + checksumFileInputStream.readFully(b); + int actualLength = b.length; if (actualLength == nameLength) { nodeName = new String(b, Charset.forName("UTF-8")); index = checksumFileInputStream.readLong(); @@ -799,7 +801,8 @@ private Long parseChecksum(FileStatus file) { checksumFileInputStream = fileContext.open(file.getPath()); int nameLength = checksumFileInputStream.readInt(); byte[] b = new byte[nameLength]; - int actualLength = checksumFileInputStream.read(b); + checksumFileInputStream.readFully(b); + int actualLength = b.length; if (actualLength == nameLength) { nodeName = new String(b, StandardCharsets.UTF_8); index = checksumFileInputStream.readLong(); @@ -938,7 +941,8 @@ public IndexedLogsMeta loadIndexedLogsMeta(Path remoteLogPath, long end, // Load UUID and make sure the UUID is correct. byte[] uuidRead = new byte[UUID_LENGTH]; - int uuidReadLen = fsDataIStream.read(uuidRead); + fsDataIStream.readFully(uuidRead); + int uuidReadLen = uuidRead.length; if (this.uuid == null) { this.uuid = createUUID(appId); } @@ -1322,7 +1326,8 @@ private byte[] loadUUIDFromLogFile(final FileContext fc, .endsWith(CHECK_SUM_FILE_SUFFIX)) { fsDataInputStream = fc.open(checkPath); byte[] b = new byte[uuid.length]; - int actual = fsDataInputStream.read(b); + fsDataInputStream.readFully(b); + int actual = b.length; if (actual != uuid.length || Arrays.equals(b, uuid)) { deleteFileWithRetries(fc, checkPath); } else if (id == null){