HBASE-24492 : Remove infinite loop from ProtobufLogReader#readNext (#1831)

Signed-off-by: Duo Zhang <zhangduo@apache.org>
This commit is contained in:
Viraj Jasani 2020-06-03 13:41:15 +05:30
parent 571ef7e811
commit b3c6af9500
No known key found for this signature in database
GPG Key ID: 3AE697641452FC5D
1 changed files with 101 additions and 101 deletions

View File

@ -329,7 +329,6 @@ public class ProtobufLogReader extends ReaderBase {
@Override @Override
protected boolean readNext(Entry entry) throws IOException { protected boolean readNext(Entry entry) throws IOException {
while (true) {
// OriginalPosition might be < 0 on local fs; if so, it is useless to us. // OriginalPosition might be < 0 on local fs; if so, it is useless to us.
long originalPosition = this.inputStream.getPos(); long originalPosition = this.inputStream.getPos();
if (trailerPresent && originalPosition > 0 && originalPosition == this.walEditsStopOffset) { if (trailerPresent && originalPosition > 0 && originalPosition == this.walEditsStopOffset) {
@ -350,23 +349,25 @@ public class ProtobufLogReader extends ReaderBase {
// available may be < 0 on local fs for instance. If so, can't depend on it. // available may be < 0 on local fs for instance. If so, can't depend on it.
available = this.inputStream.available(); available = this.inputStream.available();
if (available > 0 && available < size) { if (available > 0 && available < size) {
throw new EOFException("Available stream not enough for edit, " + throw new EOFException(
"inputStream.available()= " + this.inputStream.available() + ", " + "Available stream not enough for edit, " + "inputStream.available()= "
"entry size= " + size + " at offset = " + this.inputStream.getPos()); + this.inputStream.available() + ", " + "entry size= " + size + " at offset = "
+ this.inputStream.getPos());
} }
ProtobufUtil.mergeFrom(builder, ByteStreams.limit(this.inputStream, size), ProtobufUtil.mergeFrom(builder, ByteStreams.limit(this.inputStream, size), (int) size);
(int)size);
} catch (InvalidProtocolBufferException ipbe) { } catch (InvalidProtocolBufferException ipbe) {
resetPosition = true; resetPosition = true;
throw (EOFException) new EOFException("Invalid PB, EOF? Ignoring; originalPosition=" + throw (EOFException) new EOFException(
originalPosition + ", currentPosition=" + this.inputStream.getPos() + "Invalid PB, EOF? Ignoring; originalPosition=" + originalPosition + ", currentPosition="
", messageSize=" + size + ", currentAvailable=" + available).initCause(ipbe); + this.inputStream.getPos() + ", messageSize=" + size + ", currentAvailable="
+ available).initCause(ipbe);
} }
if (!builder.isInitialized()) { if (!builder.isInitialized()) {
// TODO: not clear if we should try to recover from corrupt PB that looks semi-legit. // TODO: not clear if we should try to recover from corrupt PB that looks semi-legit.
// If we can get the KV count, we could, theoretically, try to get next record. // If we can get the KV count, we could, theoretically, try to get next record.
throw new EOFException("Partial PB while reading WAL, " + throw new EOFException(
"probably an unexpected EOF, ignoring. current offset=" + this.inputStream.getPos()); "Partial PB while reading WAL, " + "probably an unexpected EOF, ignoring. current offset="
+ this.inputStream.getPos());
} }
WALKey walKey = builder.build(); WALKey walKey = builder.build();
entry.getKey().readFieldsFromPb(walKey, this.byteStringUncompressor); entry.getKey().readFieldsFromPb(walKey, this.byteStringUncompressor);
@ -391,31 +392,31 @@ public class ProtobufLogReader extends ReaderBase {
} catch (Throwable t) { } catch (Throwable t) {
LOG.trace("Error getting pos for error message - ignoring", t); LOG.trace("Error getting pos for error message - ignoring", t);
} }
String message = " while reading " + expectedCells + " WAL KVs; started reading at " String message =
+ posBefore + " and read up to " + posAfterStr; " while reading " + expectedCells + " WAL KVs; started reading at " + posBefore
+ " and read up to " + posAfterStr;
IOException realEofEx = extractHiddenEof(ex); IOException realEofEx = extractHiddenEof(ex);
throw (EOFException) new EOFException("EOF " + message). throw (EOFException) new EOFException("EOF " + message).
initCause(realEofEx != null ? realEofEx : ex); initCause(realEofEx != null ? realEofEx : ex);
} }
if (trailerPresent && this.inputStream.getPos() > this.walEditsStopOffset) { if (trailerPresent && this.inputStream.getPos() > this.walEditsStopOffset) {
LOG.error("Read WALTrailer while reading WALEdits. wal: " + this.path LOG.error(
+ ", inputStream.getPos(): " + this.inputStream.getPos() + ", walEditsStopOffset: " "Read WALTrailer while reading WALEdits. wal: " + this.path + ", inputStream.getPos(): "
+ this.walEditsStopOffset); + this.inputStream.getPos() + ", walEditsStopOffset: " + this.walEditsStopOffset);
throw new EOFException("Read WALTrailer while reading WALEdits"); throw new EOFException("Read WALTrailer while reading WALEdits");
} }
} catch (EOFException eof) { } catch (EOFException eof) {
// If originalPosition is < 0, it is rubbish and we cannot use it (probably local fs) // If originalPosition is < 0, it is rubbish and we cannot use it (probably local fs)
if (originalPosition < 0) { if (originalPosition < 0) {
LOG.warn("Encountered a malformed edit, but can't seek back to last good position " LOG.warn("Encountered a malformed edit, but can't seek back to last good position "
+ "because originalPosition is negative. last offset={}", + "because originalPosition is negative. last offset={}", this.inputStream.getPos(), eof);
this.inputStream.getPos(), eof);
throw eof; throw eof;
} }
// If stuck at the same place and we got and exception, lets go back at the beginning. // If stuck at the same place and we got and exception, lets go back at the beginning.
if (inputStream.getPos() == originalPosition) { if (inputStream.getPos() == originalPosition) {
if (resetPosition) { if (resetPosition) {
LOG.warn("Encountered a malformed edit, seeking to the beginning of the WAL since " + LOG.warn("Encountered a malformed edit, seeking to the beginning of the WAL since "
"current position and original position match at {}", originalPosition); + "current position and original position match at {}", originalPosition);
seekOnFs(0); seekOnFs(0);
} else { } else {
LOG.debug("Reached the end of file at position {}", originalPosition); LOG.debug("Reached the end of file at position {}", originalPosition);
@ -423,15 +424,14 @@ public class ProtobufLogReader extends ReaderBase {
} else { } else {
// Else restore our position to original location in hope that next time through we will // Else restore our position to original location in hope that next time through we will
// read successfully. // read successfully.
LOG.warn("Encountered a malformed edit, seeking back to last good position in file, " + LOG.warn("Encountered a malformed edit, seeking back to last good position in file, "
"from {} to {}", inputStream.getPos(), originalPosition, eof); + "from {} to {}", inputStream.getPos(), originalPosition, eof);
seekOnFs(originalPosition); seekOnFs(originalPosition);
} }
return false; return false;
} }
return true; return true;
} }
}
private IOException extractHiddenEof(Exception ex) { private IOException extractHiddenEof(Exception ex) {
// There are two problems we are dealing with here. Hadoop stream throws generic exception // There are two problems we are dealing with here. Hadoop stream throws generic exception