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

HBASE-15093 Replication can report incorrect size of log queue for the global source when multiwal is enabled (Ashu Pachauri)

This commit is contained in:
tedyu 2016-04-11 08:23:34 -07:00
parent e407e1e81d
commit b7502feff3
5 changed files with 25 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsReplicationSourceSource.java
View File

@ -36,7 +36,6 @@ public interface MetricsReplicationSourceSource {
public static final String SOURCE_SIZE_OF_HFILE_REFS_QUEUE = "source.sizeOfHFileRefsQueue";
void setLastShippedAge(long age);
void setSizeOfLogQueue(int size);
void incrSizeOfLogQueue(int size);
void decrSizeOfLogQueue(int size);
void incrLogEditsFiltered(long size);
@ -47,6 +46,7 @@ public interface MetricsReplicationSourceSource {
void incrLogReadInEdits(long size);
void clear();
long getLastShippedAge();
int getSizeOfLogQueue();
void incrHFilesShipped(long hfiles);
void incrSizeOfHFileRefsQueue(long size);
void decrSizeOfHFileRefsQueue(long size);

9
hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsReplicationGlobalSourceSource.java
View File

@ -64,10 +64,6 @@ public class MetricsReplicationGlobalSourceSource implements MetricsReplicationS
ageOfLastShippedOpGauge.set(age);
}
@Override public void setSizeOfLogQueue(int size) {
sizeOfLogQueueGauge.set(size);
}
@Override public void incrSizeOfLogQueue(int size) {
sizeOfLogQueueGauge.incr(size);
}
@ -121,4 +117,9 @@ public class MetricsReplicationGlobalSourceSource implements MetricsReplicationS
public void decrSizeOfHFileRefsQueue(long size) {
sizeOfHFileRefsQueueGauge.decr(size);
}
@Override
public int getSizeOfLogQueue() {
return (int)sizeOfLogQueueGauge.value();
}
}

9
hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsReplicationSourceSourceImpl.java
View File

@ -85,10 +85,6 @@ public class MetricsReplicationSourceSourceImpl implements MetricsReplicationSou
ageOfLastShippedOpGauge.set(age);
}
@Override public void setSizeOfLogQueue(int size) {
sizeOfLogQueueGauge.set(size);
}
@Override public void incrSizeOfLogQueue(int size) {
sizeOfLogQueueGauge.incr(size);
}
@ -158,4 +154,9 @@ public class MetricsReplicationSourceSourceImpl implements MetricsReplicationSou
public void decrSizeOfHFileRefsQueue(long size) {
sizeOfHFileRefsQueueGauge.decr(size);
}
@Override
public int getSizeOfLogQueue() {
return (int)sizeOfLogQueueGauge.value();
}
}

18
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java
View File

@ -89,14 +89,16 @@ public class MetricsSource {
}
/**
* Set the size of the log queue
*
* @param size the size.
* Increment size of the log queue.
*/
public void setSizeOfLogQueue(int size) {
singleSourceSource.setSizeOfLogQueue(size);
globalSourceSource.incrSizeOfLogQueue(size - lastQueueSize);
lastQueueSize = size;
public void incrSizeOfLogQueue() {
singleSourceSource.incrSizeOfLogQueue(1);
globalSourceSource.incrSizeOfLogQueue(1);
}
public void decrSizeOfLogQueue() {
singleSourceSource.decrSizeOfLogQueue(1);
globalSourceSource.decrSizeOfLogQueue(1);
}
/**
@ -186,7 +188,7 @@ public class MetricsSource {
* @return sizeOfLogQueue
*/
public int getSizeOfLogQueue() {
return this.lastQueueSize;
return singleSourceSource.getSizeOfLogQueue();
}
/**

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

@ -31,7 +31,6 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang.StringUtils;
@ -138,7 +137,6 @@ public class ReplicationSource extends Thread
private WALEntryFilter walEntryFilter;
// throttler
private ReplicationThrottler throttler;
private AtomicInteger logQueueSize = new AtomicInteger(0);
private ConcurrentHashMap<String, ReplicationSourceWorkerThread> workerThreads =
new ConcurrentHashMap<String, ReplicationSourceWorkerThread>();
@ -221,10 +219,10 @@ public class ReplicationSource extends Thread
}
}
queue.put(log);
int queueSize = logQueueSize.incrementAndGet();
this.metrics.setSizeOfLogQueue(queueSize);
this.metrics.incrSizeOfLogQueue();
// This will log a warning for each new log that gets created above the warn threshold
if (queue.size() > this.logQueueWarnThreshold) {
int queueSize = queue.size();
if (queueSize > this.logQueueWarnThreshold) {
LOG.warn("WAL group " + logPrefix + " queue size: " + queueSize
+ " exceeds value of replication.source.log.queue.warn: " + logQueueWarnThreshold);
}
@ -769,8 +767,7 @@ public class ReplicationSource extends Thread
try {
if (this.currentPath == null) {
this.currentPath = queue.poll(sleepForRetries, TimeUnit.MILLISECONDS);
int queueSize = logQueueSize.decrementAndGet();
metrics.setSizeOfLogQueue(queueSize);
metrics.decrSizeOfLogQueue();
if (this.currentPath != null) {
// For recovered queue: must use peerClusterZnode since peerId is a parsed value
manager.cleanOldLogs(this.currentPath.getName(), peerClusterZnode,