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

HBASE-27368 Do not need to throw IllegalStateException when peer is not active in ReplicationSource.initialize (#4779) 

Signed-off-by: Xin Sun <ddupgs@gmail.com>
This commit is contained in:
Duo Zhang 2022-09-15 15:27:06 +08:00 committed by GitHub
parent 00aba9ca74
commit 460bb47c04
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -421,7 +421,7 @@ public class ReplicationSource implements ReplicationSourceInterface {
return walEntryFilter; return walEntryFilter;
} }
protected final void uncaughtException(Thread t, Throwable e, ReplicationSourceManager manager, private void uncaughtException(Thread t, Throwable e, ReplicationSourceManager manager,
String peerId) { String peerId) {
OOMEChecker.exitIfOOME(e, getClass().getSimpleName()); OOMEChecker.exitIfOOME(e, getClass().getSimpleName());
LOG.error("Unexpected exception in {} currentPath={}", t.getName(), getCurrentPath(), e); LOG.error("Unexpected exception in {} currentPath={}", t.getName(), getCurrentPath(), e);
@ -539,17 +539,12 @@ public class ReplicationSource implements ReplicationSourceInterface {
} }
if (!this.isSourceActive()) { if (!this.isSourceActive()) {
// this means the server is shutting down or the source is terminated, just give up
// initializing
setSourceStartupStatus(false); setSourceStartupStatus(false);
if (Thread.currentThread().isInterrupted()) {
// If source is not running and thread is interrupted this means someone has tried to
// remove this peer.
return; return;
} }
retryStartup.set(!this.abortOnError);
throw new IllegalStateException("Source should be active.");
}
sleepMultiplier = 1; sleepMultiplier = 1;
UUID peerClusterId; UUID peerClusterId;
// delay this until we are in an asynchronous thread // delay this until we are in an asynchronous thread
@ -569,15 +564,12 @@ public class ReplicationSource implements ReplicationSourceInterface {
} }
if (!this.isSourceActive()) { if (!this.isSourceActive()) {
// this means the server is shutting down or the source is terminated, just give up
// initializing
setSourceStartupStatus(false); setSourceStartupStatus(false);
if (Thread.currentThread().isInterrupted()) {
// If source is not running and thread is interrupted this means someone has tried to
// remove this peer.
return; return;
} }
retryStartup.set(!this.abortOnError);
throw new IllegalStateException("Source should be active.");
}
LOG.info("{} queueId={} (queues={}) is replicating from cluster={} to cluster={}", logPeerId(), LOG.info("{} queueId={} (queues={}) is replicating from cluster={} to cluster={}", logPeerId(),
this.replicationQueueInfo.getQueueId(), logQueue.getNumQueues(), clusterId, peerClusterId); this.replicationQueueInfo.getQueueId(), logQueue.getNumQueues(), clusterId, peerClusterId);
initializeWALEntryFilter(peerClusterId); initializeWALEntryFilter(peerClusterId);