Skip to content

HBASE-26093 Replication is stuck due to zero length wal file in oldWALs directory #3504

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
apurtell merged 9 commits into from
Jul 23, 2021
Merged

HBASE-26093 Replication is stuck due to zero length wal file in oldWALs directory #3504

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
36818f7
HBASE-26093 Replication is stuck due to zero length wal file in oldWA…
Jul 19, 2021
5a86597
Merge branch 'master' of https://github.com/apache/hbase into HBASE-2…
Jul 19, 2021
c762357
HBASE-26093 Addressing CR comments
Jul 19, 2021
397d325
Merge branch 'master' of https://github.com/apache/hbase into HBASE-2…
Jul 20, 2021
a0d092b
HBASE-26093 Addressing review comments
Jul 20, 2021
21827b8
Merge branch 'master' of https://github.com/apache/hbase into HBASE-2…
Jul 21, 2021
7a220d6
@HBASE-26093 Addressing review comments
Jul 21, 2021
c53892b
@HBASE-26093 Addressing review comments
Jul 21, 2021
8450440
HBASE-26093 Fixing test failures
Jul 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions ...ain/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.hadoop.hbase.replication.WALEntryFilter;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.apache.hadoop.hbase.wal.WALKey;
Expand Down Expand Up @@ -274,11 +275,14 @@ private boolean handleEofException(Exception e, WALEntryBatch batch) {
// 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) {
Path head = queue.peek();
Path path = queue.peek();
try {
if (fs.getFileStatus(head).getLen() == 0) {
// head of the queue is an empty log file
LOG.warn("Forcing removal of 0 length log in queue: {}", head);
if (!fs.exists(path)) {
// There is a chance that wal has moved to oldWALs directory, so look there also.
path = AbstractFSWALProvider.getArchivedLogPath(path, conf);

@Apache9 Apache9 Jul 21, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, this is not very stable...

You need to try all the possible path, just like what we have done in WALEntryStream. So I suggest we move the main logic of WALEntryStream.getArchivedLog to AbstractFSWALProvider, as a util method, and we call this method in both places.

@shahrs87 shahrs87 Jul 21, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you @Apache9 for the feedback. I added a new util method in AbstractFSWALProvider to look into both the oldWALs path if it can't find the log in WALs directory.
After the change, there are 2 methods in AbstractFSWALProvider now to get archiveLog path:

  1. getArchivedLog This is the method added by this patch.
  2. getArchivedLogPath This was already present which looks into separate oldLogDir path based on config.

I think we can remove getArchivedLogPath method and replace all callers with getArchivedLog. WDYT ?

@Apache9 Apache9 Jul 21, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, just saw this comment. Please see my new comment below, you need to keep getArchivedLogPath, as it is used when we want to archive a wal file.
It is OK to remove the getArchivedLog in WALEntryStream.

}
if (fs.getFileStatus(path).getLen() == 0) {
LOG.warn("Forcing removal of 0 length log in queue: {}", path);
logQueue.remove(walGroupId);
currentPosition = 0;
if (batch != null) {
Expand All @@ -289,7 +293,7 @@ private boolean handleEofException(Exception e, WALEntryBatch batch) {
return true;
}
} catch (IOException ioe) {
LOG.warn("Couldn't get file length information about log " + queue.peek(), ioe);
LOG.warn("Couldn't get file length information about log " + path, ioe);
} catch (InterruptedException ie) {
LOG.trace("Interrupted while adding WAL batch to ship queue");
Thread.currentThread().interrupt();
Expand Down
46 changes: 46 additions & 0 deletions ...c/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,14 @@
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.Waiter.ExplainingPredicate;
import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
import org.apache.hadoop.hbase.replication.WALEntryFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.hadoop.hbase.wal.WAL;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
Expand Down Expand Up @@ -716,4 +718,48 @@ public void testCleanClosedWALs() throws Exception {
assertEquals(0, logQueue.getMetrics().getUncleanlyClosedWALs());
}
}

/**
* Tests that we handle EOFException properly if the wal has moved to oldWALs directory.
* @throws Exception exception
*/
@Test
public void testEOFExceptionInOldWALsDirectory() throws Exception {
assertEquals(1, logQueue.getQueueSize(fakeWalGroupId));
AbstractFSWAL abstractWAL = (AbstractFSWAL)log;

@bharathv bharathv Jul 19, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: We don't need this down cast, do we?

@shahrs87 shahrs87 Jul 19, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@bharathv It is needed. We are getting the current wal file name via AbstractFSWAL#getCurrentFileName so that we can truncate that wal file to create 0 size wal file. WAL class doesn't have this method.

Path emptyLogFile = abstractWAL.getCurrentFileName();
log.rollWriter(true);

// AsyncFSWAl and FSHLog both moves the log from WALs to oldWALs directory asynchronously.
// Wait for in flight wal close count to become 0. This makes sure that empty wal is moved to
// oldWALs directory.
Waiter.waitFor(CONF, 5000,
(Waiter.Predicate<Exception>) () -> abstractWAL.getInflightWALCloseCount() == 0);
// There will 2 logs in the queue.
assertEquals(2, logQueue.getQueueSize(fakeWalGroupId));

// Get the archived dir path for the first wal.
Path archivePath = AbstractFSWALProvider.getArchivedLogPath(emptyLogFile, CONF);
// Make sure that the wal path is not the same as archived Dir path.
assertNotEquals(emptyLogFile.toString(), archivePath.toString());
assertTrue(fs.exists(archivePath));
fs.truncate(archivePath, 0);
// make sure the size of the wal file is 0.
assertEquals(0, fs.getFileStatus(archivePath).getLen());

ReplicationSourceManager mockSourceManager = Mockito.mock(ReplicationSourceManager.class);
ReplicationSource source = Mockito.mock(ReplicationSource.class);
when(source.isPeerEnabled()).thenReturn(true);
when(mockSourceManager.getTotalBufferUsed()).thenReturn(new AtomicLong(0));

Configuration localConf = new Configuration(CONF);
localConf.setInt("replication.source.maxretriesmultiplier", 1);
localConf.setBoolean("replication.source.eof.autorecovery", true);
// Start the reader thread.
createReader(false, localConf);
// Wait for the replication queue size to be 1. This means that we have handled
// 0 length wal from oldWALs directory.
Waiter.waitFor(localConf, 10000,
(Waiter.Predicate<Exception>) () -> logQueue.getQueueSize(fakeWalGroupId) == 1);
}
}