Index: src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java (revision 1387405) +++ src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java (working copy) @@ -338,10 +338,6 @@ } } finally { try { - // if current path is null, it means we processEndOfFile hence - if (this.currentPath != null && !gotIOE) { - this.position = this.reader.getPosition(); - } if (this.reader != null) { this.reader.close(); } @@ -391,7 +387,7 @@ if (this.position != 0) { this.reader.seek(this.position); } - HLog.Entry entry = this.reader.next(this.entriesArray[currentNbEntries]); + HLog.Entry entry = readNextAndSetPosition(); while (entry != null) { WALEdit edit = entry.getEdit(); this.metrics.logEditsReadRate.inc(1); @@ -426,7 +422,7 @@ break; } try { - entry = this.reader.next(entriesArray[currentNbEntries]); + entry = readNextAndSetPosition(); } catch (IOException ie) { LOG.debug("Break on IOE: " + ie.getMessage()); break; @@ -440,6 +436,16 @@ return seenEntries == 0 && processEndOfFile(); } + private HLog.Entry readNextAndSetPosition() throws IOException { + HLog.Entry entry = this.reader.next(entriesArray[currentNbEntries]); + // Store the position so that in the future the reader can start + // reading from here. If the above call to next() throws an + // exception, the position won't be changed and retry will happen + // from the last known good position + this.position = this.reader.getPosition(); + return entry; + } + private void connectToPeers() { // Connect to peer cluster first, unless we have to stop while (this.isActive() && this.currentPeers.size() == 0) {