Skip to content

HBASE-24877 Add option to avoid aborting RS process upon uncaught exc… #2255

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
wchevreuil merged 7 commits into from
Sep 14, 2020
Merged

HBASE-24877 Add option to avoid aborting RS process upon uncaught exc… #2255

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5a0adaf
HBASE-24877 Add option to avoid aborting RS process upon uncaught exc…
wchevreuil Aug 12, 2020
3fb4dce
Addressing Viraj comments
wchevreuil Aug 14, 2020
8af3041
Addressing Duo's comments.
wchevreuil Aug 24, 2020
84994f9
addressing some nits and the loop control
wchevreuil Aug 24, 2020
e716047
Changed to MutableBoolean and addressed latest checkstyles
wchevreuil Aug 25, 2020
2fcf714
switched from banned version
wchevreuil Aug 25, 2020
16db51e
Additional checks on the loop control, for the tests that are note co…
wchevreuil Sep 4, 2020
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
57 changes: 51 additions & 6 deletions ...ver/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;
import org.apache.commons.lang3.StringUtils;
Expand Down Expand Up @@ -119,6 +120,17 @@ public class ReplicationSource implements ReplicationSourceInterface {
private int logQueueWarnThreshold;
// ReplicationEndpoint which will handle the actual replication
private volatile ReplicationEndpoint replicationEndpoint;
//Flag that signalizes uncaught error happening while starting up the source
// and a retry should be attempted
private final AtomicBoolean retryStartup = new AtomicBoolean(false);
//This is needed for the startup loop to identify when there's already
// an initialization happening (but not finished yet),
// so that it doesn't try submit another initialize thread.
// NOTE: this should only be set to false at the end of initialize method, prior to return.
private final AtomicBoolean startupOngoing = new AtomicBoolean(false);

private boolean abortOnError;


/**
* A filter (or a chain of filters) for WAL entries; filters out edits.
Expand Down Expand Up @@ -217,6 +229,10 @@ public void init(Configuration conf, FileSystem fs, ReplicationSourceManager man
this.throttler = new ReplicationThrottler((double) currentBandwidth / 10.0);
this.totalBufferUsed = manager.getTotalBufferUsed();
this.walFileLengthProvider = walFileLengthProvider;

this.abortOnError = this.conf.getBoolean("replication.source.regionserver.abort",
true);

LOG.info("queueId={}, ReplicationSource: {}, currentBandwidth={}", queueId,
replicationPeer.getId(), this.currentBandwidth);
}
Expand Down Expand Up @@ -373,7 +389,21 @@ private void tryStartNewShipper(String walGroupId, PriorityBlockingQueue<Path> q
Threads.setDaemonThreadRunning(
walReader, Thread.currentThread().getName()
+ ".replicationSource.wal-reader." + walGroupId + "," + queueId,
this::uncaughtException);
(t,e) -> {

@Apache9 Apache9 Aug 19, 2020

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.

So here it is for wal reader. I think refreshSources and retry is an acceptable way. Then let's just test the abortOnError flag here? If it is true, we will call uncaughtException, otherwise we will try to refresh the replication source.

this.uncaughtException(t,e);
while (true) {
try {
LOG.info("Refreshing replication sources now due " +
"to previous error on reader thread.");
this.manager.refreshSources(this.getPeerId());
break;
} catch (IOException e1) {
LOG.error("Replication sources refresh failed.", e1);
sleepForRetries("Sleeping before try refreshing sources again",
maxRetriesMultiplier);
}
}
});
worker.setWALReader(walReader);
worker.startup(this::uncaughtException);
return worker;
Expand Down Expand Up @@ -454,7 +484,9 @@ protected final void uncaughtException(Thread t, Throwable e) {
RSRpcServices.exitIfOOME(e);
LOG.error("Unexpected exception in {} currentPath={}",
t.getName(), getCurrentPath(), e);
server.abort("Unexpected exception in " + t.getName(), e);
if(abortOnError){
server.abort("Unexpected exception in " + t.getName(), e);
}
}

@Override
Expand Down Expand Up @@ -549,6 +581,7 @@ private void initialize() {
}

if (!this.isSourceActive()) {
this.startupOngoing.set(false);
return;
}

Expand All @@ -571,6 +604,7 @@ private void initialize() {
}

if(!this.isSourceActive()) {
this.startupOngoing.set(false);
return;
}
LOG.info("{} Source: {}, is now replicating from cluster: {}; to peer cluster: {};",
Expand All @@ -583,16 +617,27 @@ private void initialize() {
PriorityBlockingQueue<Path> queue = entry.getValue();
tryStartNewShipper(walGroupId, queue);
}
this.startupOngoing.set(false);
}

@Override
public void startup() {
// mark we are running now
this.sourceRunning = true;
initThread = new Thread(this::initialize);
Threads.setDaemonThreadRunning(initThread,
Thread.currentThread().getName() + ".replicationSource," + this.queueId,
this::uncaughtException);
this.retryStartup.set(true);

@Apache9 Apache9 Aug 19, 2020

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.

This flag is only used in this method? Let's use a local var instead of a class member field?

@wchevreuil wchevreuil Aug 21, 2020

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.

Yeah, originally I was also referring it on uncaughtException, but it was not actually needed.

do {
if(retryStartup.get()) {
retryStartup.set(false);
startupOngoing.set(true);
Comment thread
wchevreuil marked this conversation as resolved.
initThread = new Thread(this::initialize);
Threads.setDaemonThreadRunning(initThread,
Thread.currentThread().getName() + ".replicationSource," + this.queueId,
(t,e) -> {
uncaughtException(t, e);
retryStartup.set(true);

@virajjasani virajjasani Aug 13, 2020

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.

I hope if we encounter uncaughtException here, we want to retry the loop again.
If so, shall we also add startupOngoing.set(true); here explicitely? Just in case if it is set to false in initialize()?

@wchevreuil wchevreuil Aug 14, 2020

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.

I hope if we encounter uncaughtException here, we want to retry the loop again.

Yes, we want to keep trying until we succeed.

If so, shall we also add startupOngoing.set(true); here explicitely? Just in case if it is set to false in initialize()?

The only cases startupOngoing would had been set to false in initialize is if it completes fine without any uncaught exception, so adding it here would be redundant.

@virajjasani virajjasani Aug 14, 2020

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.

Yeah right, it is not set to false if it completes successfully, anyone touching the same code in future should realize this.
Although not a strong point but If you don't mind, maybe we can comment here indicating startupOngoing is expected to be true when we are here handling Exception.

@wchevreuil wchevreuil Aug 14, 2020

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.

Addressed the nits and added some comments to the two flags.

});
}
} while (startupOngoing.get());
}

@Override
Expand Down
17 changes: 16 additions & 1 deletion .../main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,22 @@ private boolean updateLogPosition(WALEntryBatch batch) {
public void startup(UncaughtExceptionHandler handler) {
String name = Thread.currentThread().getName();
Threads.setDaemonThreadRunning(this,
name + ".replicationSource.shipper" + walGroupId + "," + source.getQueueId(), handler);
name + ".replicationSource.shipper" + walGroupId + "," + source.getQueueId(),
(t,e) -> {

@Apache9 Apache9 Aug 19, 2020

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.

OK, the code is almost the same... Then I think we could move the logic into uncaughtException method? If abortOnError is true, we about, otherwise we will try to refresh the source.

@wchevreuil wchevreuil Aug 24, 2020

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.

Moved it to uncaughtException. This is true when handling potential errors on both the reader and shipper threads, but uncaughtException is also used when handling issues from ReplicationSource.initialize method, which might blow before reader/shipper is even started. So needed to add extra check in uncaughtException to decide when to invoke refreshSources.

handler.uncaughtException(t, e);
while (true) {
try {
LOG.info("Refreshing replication sources now due " +
"to previous error on shipper thread.");
this.source.getSourceManager().refreshSources(this.source.getPeerId());
break;
} catch (IOException e1) {
LOG.error("Replication sources refresh failed.", e1);
sleepForRetries("Sleeping before try refreshing sources again",
sleepForRetries, 1, maxRetriesMultiplier);
}
}
});
}

Path getCurrentPath() {
Expand Down
91 changes: 91 additions & 0 deletions ...src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.replication.regionserver;
import static org.apache.hadoop.hbase.wal.AbstractFSWALProvider.META_WAL_PROVIDER_ID;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
Expand Down Expand Up @@ -396,6 +397,25 @@ protected void doStop() {
}
}

/**
* Deadend Endpoint. Does nothing.
*/
public static class FaultyReplicationEndpoint extends DoNothingReplicationEndpoint {

static int count = 0;

@Override
public synchronized UUID getPeerUUID() {
if(count==0) {
count++;
throw new RuntimeException();
} else {
return super.getPeerUUID();
}
}

}

/**
* Test HBASE-20497
* Moved here from TestReplicationSource because doesn't need cluster.
Expand Down Expand Up @@ -423,5 +443,76 @@ public void testRecoveredReplicationSourceShipperGetPosition() throws Exception
new RecoveredReplicationSourceShipper(conf, walGroupId, queue, source, storage);
assertEquals(1001L, shipper.getStartPosition());
}

/**
* Test ReplicationSource retries startup once an uncaught exception happens
* during initialization and <b>eplication.source.regionserver.abort</b> is set to false.
*/
@Test
public void testAbortFalseOnError() throws IOException {
ReplicationSource rs = new ReplicationSource();
Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
conf.setInt("replication.source.maxretriesmultiplier", 1);
conf.setBoolean("replication.source.regionserver.abort", false);
ReplicationPeer mockPeer = Mockito.mock(ReplicationPeer.class);
Mockito.when(mockPeer.getConfiguration()).thenReturn(conf);
Mockito.when(mockPeer.getPeerBandwidth()).thenReturn(0L);
ReplicationPeerConfig peerConfig = Mockito.mock(ReplicationPeerConfig.class);
Mockito.when(peerConfig.getReplicationEndpointImpl()).
thenReturn(FaultyReplicationEndpoint.class.getName());
Mockito.when(mockPeer.getPeerConfig()).thenReturn(peerConfig);
ReplicationSourceManager manager = Mockito.mock(ReplicationSourceManager.class);
Mockito.when(manager.getTotalBufferUsed()).thenReturn(new AtomicLong());
String queueId = "qid";
RegionServerServices rss =
TEST_UTIL.createMockRegionServerService(ServerName.parseServerName("a.b.c,1,1"));
rs.init(conf, null, manager, null, mockPeer, rss, queueId, null,
p -> OptionalLong.empty(), new MetricsSource(queueId));
try {
rs.startup();
assertTrue(rs.isSourceActive());
assertEquals(0, rs.getSourceMetrics().getSizeOfLogQueue());
rs.enqueueLog(new Path("a.1" + META_WAL_PROVIDER_ID));
assertEquals(0, rs.getSourceMetrics().getSizeOfLogQueue());
rs.enqueueLog(new Path("a.1"));
assertEquals(1, rs.getSourceMetrics().getSizeOfLogQueue());
} finally {
rs.terminate("Done");
rss.stop("Done");
}
}

/**
* Test ReplicationSource retries startup once an uncaught exception happens
* during initialization and <b>eplication.source.regionserver.abort</b> is set to false.
*/
@Test
public void testAbortTrueOnError() throws IOException {
ReplicationSource rs = new ReplicationSource();
Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
conf.setInt("replication.source.maxretriesmultiplier", 1);
ReplicationPeer mockPeer = Mockito.mock(ReplicationPeer.class);
Mockito.when(mockPeer.getConfiguration()).thenReturn(conf);
Mockito.when(mockPeer.getPeerBandwidth()).thenReturn(0L);
ReplicationPeerConfig peerConfig = Mockito.mock(ReplicationPeerConfig.class);
Mockito.when(peerConfig.getReplicationEndpointImpl()).
thenReturn(FaultyReplicationEndpoint.class.getName());
Mockito.when(mockPeer.getPeerConfig()).thenReturn(peerConfig);
ReplicationSourceManager manager = Mockito.mock(ReplicationSourceManager.class);
Mockito.when(manager.getTotalBufferUsed()).thenReturn(new AtomicLong());
String queueId = "qid";
RegionServerServices rss =
TEST_UTIL.createMockRegionServerService(ServerName.parseServerName("a.b.c,1,1"));
rs.init(conf, null, manager, null, mockPeer, rss, queueId, null,
p -> OptionalLong.empty(), new MetricsSource(queueId));
try {
rs.startup();
assertFalse(rs.isSourceActive());
assertTrue(rss.isAborted());
} finally {
rs.terminate("Done");
rss.stop("Done");
}
}
}