sandeepvinayak commented on a change in pull request #2987:
URL: https://github.com/apache/hbase/pull/2987#discussion_r583097076
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
##########
@@ -245,22 +270,37 @@ private void handleEmptyWALEntryBatch() throws
InterruptedException {
}
/**
- * if we get an EOF due to a zero-length log, and there are other logs in
queue
- * (highly likely we've closed the current log), and autorecovery is
- * enabled, then dump the log
+ * This is to handle the EOFException from the WAL entry stream.
EOFException should
+ * be handled carefully because there are chances of data loss because of
never replicating
+ * the data. Thus we should always try to ship existing batch of entries
here.
+ * If there was only one log in the queue before EOF, we ship the empty
batch here
+ * and since reader is still active, in the next iteration of reader we will
+ * stop the reader.
+ * If there was more than one log in the queue before EOF, we ship the
existing batch
+ * and reset the wal patch and position to the log with EOF, so shipper can
remove
+ * logs from replication queue
* @return true only the IOE can be handled
*/
- private boolean handleEofException(IOException e) {
+ private boolean handleEofException(IOException e, WALEntryBatch batch)
+ throws InterruptedException {
PriorityBlockingQueue<Path> queue = logQueue.getQueue(walGroupId);
// Dump the log even if logQueue size is 1 if the source is from recovered
Source
// since we don't add current log to recovered source queue so it is safe
to remove.
- if ((e instanceof EOFException || e.getCause() instanceof EOFException) &&
- (source.isRecovered() || queue.size() > 1) && this.eofAutoRecovery) {
+ if ((e instanceof EOFException || e.getCause() instanceof EOFException)
+ && (source.isRecovered() || queue.size() > 1)
+ && this.eofAutoRecovery) {
try {
if (fs.getFileStatus(queue.peek()).getLen() == 0) {
Review comment:
@apurtell if this is the case where
`fs.getFileStatus(queue.peek()).getLen() == 0`
it will give the same result because the top of the queue is not going to be
removed from anywhere except this handling of an exception. Let me know if it
makes sense.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
