wchevreuil commented on a change in pull request #934: HBASE-23568: Improve
Threading of Replication
URL: https://github.com/apache/hbase/pull/934#discussion_r376323949
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
##########
@@ -314,31 +328,59 @@ private void initializeWALEntryFilter(UUID
peerClusterId) {
this.walEntryFilter = new ChainWALEntryFilter(filters);
}
- private void tryStartNewShipper(String walGroupId,
PriorityBlockingQueue<Path> queue) {
- workerThreads.compute(walGroupId, (key, value) -> {
- if (value != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "{} Someone has beat us to start a worker thread for wal group
{}",
- logPeerId(), key);
- }
- return value;
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("{} Starting up worker for wal group {}", logPeerId(),
key);
+ /**
+ * Synchronized method so that only one item is inserted at a time. Should
be the only place that
+ * insertions are performed on the workerThreads data structure.
+ *
+ * @param walGroupId The WAL group ID
+ * @param queue The queue of paths to process
+ */
+ private synchronized void tryStartNewShipper(final String walGroupId,
+ final PriorityBlockingQueue<Path> queue) {
+
+ if (workerThreads.containsKey(walGroupId)) {
Review comment:
Could we keep debugging what seems a race condition, just like we used to do
before?
`LOG.debug( * @param walGroupId The WAL group ID
"{} Someone has beat us to start a worker thread for wal group
{}", * @param queue The queue of paths to process
logPeerId(), key);`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
