belugabehr commented on a change in pull request #934: HBASE-23568: Improve
Threading of Replication
URL: https://github.com/apache/hbase/pull/934#discussion_r375864354
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
##########
@@ -304,31 +318,56 @@ 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)) {
+ return;
+ }
+
+ LOG.debug("{} Register worker for wallGroupID {}", logPeerId(),
walGroupId);
Review comment:
Not very performance sensitive code, gets called a couple of times on
initialization, but will add since you brought it to my attention.
----------------------------------------------------------------
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
