Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-25992 Polish the ReplicationSourceWALReader code for 2.x after HBASE-25596 (#3376) 

Signed-off-by: Yulin Niu <niuyulin@apache.org>
This commit is contained in:
Duo Zhang 2021-06-20 16:32:42 +08:00
parent 3b58919246
commit c800e6c825
3 changed files with 94 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
View File

@ -122,16 +122,14 @@ class ReplicationSourceWALReader extends Thread {
@Override
public void run() {
int sleepMultiplier = 1;
while (isReaderRunning()) { // we only loop back here if something fatal happened to our stream
WALEntryBatch batch = null;
WALEntryStream entryStream = null;
try {
// we only loop back here if something fatal happened to our stream
while (isReaderRunning()) {
try {
entryStream =
new WALEntryStream(logQueue, conf, currentPosition, source.getWALFileLengthProvider(),
source.getServerWALsBelongTo(), source.getSourceMetrics(), walGroupId);
try (WALEntryStream entryStream =
new WALEntryStream(logQueue, conf, currentPosition,
source.getWALFileLengthProvider(), source.getServerWALsBelongTo(),
source.getSourceMetrics(), walGroupId)) {
while (isReaderRunning()) { // loop here to keep reusing stream while we can
batch = null;
if (!source.isPeerEnabled()) {
Threads.sleep(sleepForRetries);
continue;
@ -139,49 +137,37 @@ class ReplicationSourceWALReader extends Thread {
if (!checkQuota()) {
continue;
}
batch = createBatch(entryStream);
batch = readWALEntries(entryStream, batch);
currentPosition = entryStream.getPosition();
batch = tryAdvanceStreamAndCreateWALBatch(entryStream);
if (batch == null) {
// either the queue have no WAL to read
// or got no new entries (didn't advance position in WAL)
// got no entries and didn't advance position in WAL
handleEmptyWALEntryBatch();
entryStream.reset(); // reuse stream
} else {
addBatchToShippingQueue(batch);
continue;
}
// if we have already switched a file, skip reading and put it directly to the ship queue
if (!batch.isEndOfFile()) {
readWALEntries(entryStream, batch);
currentPosition = entryStream.getPosition();
}
} catch (WALEntryFilterRetryableException | IOException e) { // stream related
if (handleEofException(e, batch)) {
// need to propagate the batch even it has no entries since it may carry the last
// sequence id information for serial replication.
LOG.debug("Read {} WAL entries eligible for replication", batch.getNbEntries());
entryBatchQueue.put(batch);
sleepMultiplier = 1;
} else {
LOG.warn("Failed to read stream of replication entries "
+ "or replication filter is recovering", e);
}
} catch (IOException e) { // stream related
if (!handleEofException(e, batch)) {
LOG.warn("Failed to read stream of replication entries", e);
if (sleepMultiplier < maxRetriesMultiplier) {
sleepMultiplier ++;
}
Threads.sleep(sleepForRetries * sleepMultiplier);
}
} catch (InterruptedException e) {
LOG.trace("Interrupted while sleeping between WAL reads");
LOG.trace("Interrupted while sleeping between WAL reads or adding WAL batch to ship queue");
Thread.currentThread().interrupt();
} finally {
entryStream.close();
}
}
} catch (IOException e) {
if (sleepMultiplier < maxRetriesMultiplier) {
LOG.debug("Failed to read stream of replication entries: ", e);
sleepMultiplier++;
} else {
LOG.error("Failed to read stream of replication entries", e);
}
Threads.sleep(sleepForRetries * sleepMultiplier);
} catch (InterruptedException e) {
LOG.trace("Interrupted while sleeping between WAL reads");
Thread.currentThread().interrupt();
}
}
// returns true if we reach the size limit for batch, i.e, we need to finish the batch and return.
@ -211,29 +197,10 @@ class ReplicationSourceWALReader extends Thread {
// We need to get the WALEntryBatch from the caller so we can add entries in there
// This is required in case there is any exception in while reading entries
// we do want to loss the existing entries in the batch
protected WALEntryBatch readWALEntries(WALEntryStream entryStream,
WALEntryBatch batch) throws IOException, InterruptedException {
// we do not want to loss the existing entries in the batch
protected void readWALEntries(WALEntryStream entryStream, WALEntryBatch batch)
throws IOException, InterruptedException {
Path currentPath = entryStream.getCurrentPath();
if (!entryStream.hasNext()) {
// check whether we have switched a file
if (currentPath != null && switched(entryStream, currentPath)) {
return WALEntryBatch.endOfFile(currentPath);
} else {
// This would mean either no more files in the queue
// or there is no new data yet on the current wal
return null;
}
}
if (currentPath != null) {
if (switched(entryStream, currentPath)) {
return WALEntryBatch.endOfFile(currentPath);
}
} else {
// when reading from the entry stream first time we will enter here
currentPath = entryStream.getCurrentPath();
}
batch.setLastWalPath(currentPath);
for (;;) {
Entry entry = entryStream.next();
batch.setLastWalPosition(entryStream.getPosition());
@ -253,7 +220,6 @@ class ReplicationSourceWALReader extends Thread {
break;
}
}
return batch;
}
private void handleEmptyWALEntryBatch() throws InterruptedException {
@ -270,6 +236,25 @@ class ReplicationSourceWALReader extends Thread {
}
}
private WALEntryBatch tryAdvanceStreamAndCreateWALBatch(WALEntryStream entryStream)
throws IOException {
Path currentPath = entryStream.getCurrentPath();
if (!entryStream.hasNext()) {
// check whether we have switched a file
if (currentPath != null && switched(entryStream, currentPath)) {
return WALEntryBatch.endOfFile(currentPath);
} else {
return null;
}
}
if (currentPath != null) {
if (switched(entryStream, currentPath)) {
return WALEntryBatch.endOfFile(currentPath);
}
}
return createBatch(entryStream);
}
/**
* 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
@ -277,19 +262,18 @@ class ReplicationSourceWALReader extends Thread {
* 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.
* <p/>
* 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(Exception e, WALEntryBatch batch)
throws InterruptedException {
private boolean handleEofException(Exception e, WALEntryBatch batch) {
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) {
Path head = queue.peek();
try {
if (fs.getFileStatus(head).getLen() == 0) {
@ -297,16 +281,18 @@ class ReplicationSourceWALReader extends Thread {
LOG.warn("Forcing removal of 0 length log in queue: {}", head);
logQueue.remove(walGroupId);
currentPosition = 0;
// After we removed the WAL from the queue, we should
// try shipping the existing batch of entries and set the wal position
// and path to the wal just dequeued to correctly remove logs from the zk
batch.setLastWalPath(head);
batch.setLastWalPosition(currentPosition);
if (batch != null) {
// After we removed the WAL from the queue, we should try shipping the existing batch of
// entries
addBatchToShippingQueue(batch);
}
return true;
}
} catch (IOException ioe) {
LOG.warn("Couldn't get file length information about log " + queue.peek(), ioe);
} catch (InterruptedException ie) {
LOG.trace("Interrupted while adding WAL batch to ship queue");
Thread.currentThread().interrupt();
}
}
return false;
@ -316,10 +302,8 @@ class ReplicationSourceWALReader extends Thread {
* Update the batch try to ship and return true if shipped
* @param batch Batch of entries to ship
* @throws InterruptedException throws interrupted exception
* @throws IOException throws io exception from stream
*/
private void addBatchToShippingQueue(WALEntryBatch batch)
throws InterruptedException, IOException {
private void addBatchToShippingQueue(WALEntryBatch batch) throws InterruptedException {
// need to propagate the batch even it has no entries since it may carry the last
// sequence id information for serial replication.
LOG.debug("Read {} WAL entries eligible for replication", batch.getNbEntries());
@ -348,7 +332,7 @@ class ReplicationSourceWALReader extends Thread {
return true;
}
protected final WALEntryBatch createBatch(WALEntryStream entryStream) {
private WALEntryBatch createBatch(WALEntryStream entryStream) {
return new WALEntryBatch(replicationBatchCountCapacity, entryStream.getCurrentPath());
}

20
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/SerialReplicationSourceWALReader.java
View File

@ -50,27 +50,10 @@ public class SerialReplicationSourceWALReader extends ReplicationSourceWALReader
}
@Override
protected WALEntryBatch readWALEntries(WALEntryStream entryStream, WALEntryBatch batch)
protected void readWALEntries(WALEntryStream entryStream, WALEntryBatch batch)
throws IOException, InterruptedException {
Path currentPath = entryStream.getCurrentPath();
if (!entryStream.hasNext()) {
// check whether we have switched a file
if (currentPath != null && switched(entryStream, currentPath)) {
return WALEntryBatch.endOfFile(currentPath);
} else {
return null;
}
}
if (currentPath != null) {
if (switched(entryStream, currentPath)) {
return WALEntryBatch.endOfFile(currentPath);
}
} else {
// when reading from the entry stream first time we will enter here
currentPath = entryStream.getCurrentPath();
}
long positionBefore = entryStream.getPosition();
batch = createBatch(entryStream);
for (;;) {
Entry entry = entryStream.peek();
boolean doFiltering = true;
@ -122,7 +105,6 @@ public class SerialReplicationSourceWALReader extends ReplicationSourceWALReader
break;
}
}
return batch;
}
private void removeEntryFromStream(WALEntryStream entryStream, WALEntryBatch batch)

34
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEmptyWALRecovery.java
View File

@ -47,14 +47,14 @@ import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@Category
({ ReplicationTests.class, LargeTests.class }) public class TestReplicationEmptyWALRecovery
extends TestReplicationBase {
@Category({ ReplicationTests.class, LargeTests.class })
public class TestReplicationEmptyWALRecovery extends TestReplicationBase {
MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
static final RegionInfo info = RegionInfoBuilder.newBuilder(tableName).build();
NavigableMap<byte[], Integer> scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
@ClassRule public static final HBaseClassTestRule CLASS_RULE =
@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestReplicationEmptyWALRecovery.class);
@Before
@ -151,10 +151,9 @@ import org.junit.experimental.categories.Category;
}
/**
* Test empty WAL along with non empty WALs in the same batch. This test is to make sure
* when we see the empty and handle the EOF exception, we are able to existing the previous
* batch of entries without loosing it. This test also tests the number of batches shipped
*
* Test empty WAL along with non empty WALs in the same batch. This test is to make sure when we
* see the empty and handle the EOF exception, we are able to ship the previous batch of entries
* without loosing it. This test also tests the number of batches shipped
* @throws Exception throws any exception
*/
@Test
@ -174,7 +173,6 @@ import org.junit.experimental.categories.Category;
Path currentWalPath = AbstractFSWALProvider.getCurrentFileName(wal);
appendEntriesToWal(numOfEntriesToReplicate, wal);
wal.rollWriter();
String walGroupId = AbstractFSWALProvider.getWALPrefixFromWALName(currentWalPath.getName());
Path emptyWalPath = new Path(UTIL1.getDefaultRootDirPath(), walGroupId + "." + ts);
UTIL1.getTestFileSystem().create(emptyWalPath).close();
@ -183,10 +181,10 @@ import org.junit.experimental.categories.Category;
injectEmptyWAL(numRs, emptyWalPaths);
// There should be three WALs in queue
// 1. empty WAL
// 2. non empty WAL
// 1. non empty WAL
// 2. empty WAL
// 3. live WAL
//verifyNumberOfLogsInQueue(3, numRs);
verifyNumberOfLogsInQueue(3, numRs);
hbaseAdmin.enableReplicationPeer(PEER_ID2);
// ReplicationSource should advance past the empty wal, or else the test will fail
waitForLogAdvance(numRs);
@ -209,10 +207,9 @@ import org.junit.experimental.categories.Category;
}
/**
* Test empty WAL along with non empty WALs in the same batch. This test is to make sure
* when we see the empty WAL and handle the EOF exception, we are able to proceed
* with next batch and replicate it properly without missing data.
*
* Test empty WAL along with non empty WALs in the same batch. This test is to make sure when we
* see the empty WAL and handle the EOF exception, we are able to proceed with next batch and
* replicate it properly without missing data.
* @throws Exception throws any exception
*/
@Test
@ -265,9 +262,8 @@ import org.junit.experimental.categories.Category;
}
/**
* This test make sure we replicate all the enties from the non empty WALs which
* are surrounding the empty WALs
*
* This test make sure we replicate all the enties from the non empty WALs which are surrounding
* the empty WALs
* @throws Exception throws exception
*/
@Test