You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zg...@apache.org on 2020/07/21 01:20:47 UTC
[hbase] branch HBASE-24666 updated (9e8c930 -> fae9f0c)
This is an automated email from the ASF dual-hosted git repository.
zghao pushed a change to branch HBASE-24666
in repository https://gitbox.apache.org/repos/asf/hbase.git.
omit 9e8c930 HBASE-24681 Remove the cache walsById/walsByIdRecoveredQueues from ReplicationSourceManager (#2019)
omit 1005379 HBASE-24682 Refactor ReplicationSource#addHFileRefs method: move it to ReplicationSourceManager (#2020)
add d132c4b HBASE-24693 regioninfo#isLast() should point to EMPTY_END_ROW (readability improvement)
add 70cfe25 HBASE-24721: rename_rsgroup overwriting the existing rsgroup
add 3e709c6 HBASE-24720: Meta replicas not cleaned when disabled (#2057)
add 927df47 HBASE-24487 Add 2.3 Documentation to the website (#2060)
add ee226de6 HBASE-24566 Add 2.3.0 to the downloads page (#2061)
add 37a2994 HBASE-24615 MutableRangeHistogram#updateSnapshotRangeMetrics doesn't calculate the distribution for last bucket
add 1360bee HBASE-24705 MetaFixer#fixHoles() does not include the case for read replicas (i.e, replica regions are not created) (#2062)
add 2505c77 HBASE-24581 Skip compaction request/check for replica regions at the early stage. (#1986)
add a4e5ccc HBASE-24742 Improve performance of SKIP vs SEEK logic.
add 16f306b HBASE-24718 : Generic NamedQueue framework for multiple use-cases (Refactor SlowLog responses)
add 6ae2fdd HBASE-24746 The sort icons overlap the col name in master UI
add 1781185 HBASE-24675: On Master restart all servers are assigned to default rsgroup
add bd42c75 HBASE-24747 Log an ERROR if HBaseSaslRpcServer initialisation fails with an uncaught exception (#2079)
add 4b3ef81 HBASE-24722 Update commands with unintentional return values (#2058)
add 9b02a26 HBASE-24698 Turn OFF Canary WebUI as default (#2037)
add 9b32943 HBASE-24709 Support MoveCostFunction use a lower multiplier in offpeak hours
add f73b908 HBASE-24748 Add hbase.master.balancer.stochastic.moveCost.offpeak to doc as support dynamically change
add a857c6d HBASE-24714 Error message is displayed in UI of table's compaction state if any region of that table is not open
add 7ebc617 HBASE-24658 Update PolicyBasedChaosMonkey to handle uncaught exceptions
add 6cf013d HBASE-24743 Reject to add a peer which replicate to itself earlier (#2071)
new bb9cae1 HBASE-24682 Refactor ReplicationSource#addHFileRefs method: move it to ReplicationSourceManager (#2020)
new fae9f0c HBASE-24681 Remove the cache walsById/walsByIdRecoveredQueues from ReplicationSourceManager (#2019)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (9e8c930)
\
N -- N -- N refs/heads/HBASE-24666 (fae9f0c)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/hadoop/hbase/client/RegionInfo.java | 2 +-
.../hadoop/hbase/client/RegionReplicaUtil.java | 5 +-
.../hbase/shaded/protobuf/RequestConverter.java | 5 +
.../java/org/apache/hadoop/hbase/HConstants.java | 2 +-
hbase-common/src/main/resources/hbase-default.xml | 12 +
.../hadoop/metrics2/lib/MutableRangeHistogram.java | 4 +-
.../metrics2/lib/TestMutableRangeHistogram.java | 90 +++++++
.../chaos/monkies/PolicyBasedChaosMonkey.java | 72 +++---
.../src/main/protobuf/server/region/Admin.proto | 6 +
.../org/apache/hadoop/hbase/ipc/RpcServer.java | 28 +--
.../hadoop/hbase/ipc/RpcServerInterface.java | 10 +-
.../hadoop/hbase/ipc/ServerRpcConnection.java | 8 +-
.../org/apache/hadoop/hbase/master/HMaster.java | 2 +-
.../hadoop/hbase/master/MasterMetaBootstrap.java | 4 -
.../org/apache/hadoop/hbase/master/MetaFixer.java | 32 ++-
.../apache/hadoop/hbase/master/ServerManager.java | 6 +-
.../master/balancer/StochasticLoadBalancer.java | 19 +-
.../master/procedure/CreateTableProcedure.java | 2 +-
.../master/procedure/EnableTableProcedure.java | 2 +-
.../master/replication/ReplicationPeerManager.java | 50 +++-
.../DisruptorExceptionHandler.java | 2 +-
.../hadoop/hbase/namequeues/LogEventHandler.java | 130 ++++++++++
.../slowlog => namequeues}/LogHandlerUtils.java | 6 +-
.../hadoop/hbase/namequeues/NamedQueuePayload.java | 39 ++-
.../NamedQueueRecorder.java} | 120 ++++------
.../hadoop/hbase/namequeues/NamedQueueService.java | 69 ++++++
.../slowlog => namequeues}/RingBufferEnvelope.java | 22 +-
.../slowlog => namequeues}/RpcLogDetails.java | 5 +-
.../hbase/namequeues/SlowLogPersistentService.java | 98 ++++++++
.../SlowLogTableOpsChore.java | 12 +-
.../hbase/namequeues/impl/SlowLogQueueService.java | 264 ++++++++++++++++++++
.../namequeues/request/NamedQueueGetRequest.java | 65 +++++
.../namequeues/response/NamedQueueGetResponse.java | 61 +++++
.../hadoop/hbase/regionserver/HRegionServer.java | 36 ++-
.../hadoop/hbase/regionserver/RSRpcServices.java | 52 ++--
.../hadoop/hbase/regionserver/StoreScanner.java | 5 +-
.../regionserver/slowlog/LogEventHandler.java | 266 ---------------------
.../regionserver/ReplicationSource.java | 10 -
.../hbase/rsgroup/RSGroupInfoManagerImpl.java | 15 +-
.../org/apache/hadoop/hbase/tool/CanaryTool.java | 17 +-
.../main/resources/hbase-webapps/master/table.jsp | 14 +-
.../resources/hbase-webapps/static/css/hbase.css | 1 +
.../apache/hadoop/hbase/HBaseTestingUtility.java | 13 +
.../hbase/client/TestAsyncReplicationAdminApi.java | 6 +-
.../hbase/client/TestMetaWithReplicasBasic.java | 25 +-
.../TestReplicationAdminForSyncReplication.java | 2 +-
.../apache/hadoop/hbase/master/TestMetaFixer.java | 37 ++-
.../master/assignment/TestRegionReplicaSplit.java | 14 +-
.../balancer/TestStochasticLoadBalancer.java | 21 ++
.../TestNamedQueueRecorder.java} | 216 ++++++++++-------
.../TestSlowLogAccessor.java | 56 +++--
.../replication/SerialReplicationTestBase.java | 5 +
.../hbase/replication/TestMasterReplication.java | 42 +---
.../hbase/replication/TestReplicationEndpoint.java | 5 +
.../TestRaceWhenCreatingReplicationSource.java | 5 +
.../hadoop/hbase/rsgroup/TestRSGroupsAdmin1.java | 52 ++++
.../TestHBaseFsckCleanReplicationBarriers.java | 5 +-
.../src/main/ruby/shell/commands/balance_switch.rb | 2 +-
.../src/main/ruby/shell/commands/balancer.rb | 4 +-
.../ruby/shell/commands/catalogjanitor_enabled.rb | 4 +-
.../ruby/shell/commands/catalogjanitor_switch.rb | 4 +-
.../ruby/shell/commands/cleaner_chore_enabled.rb | 4 +-
.../ruby/shell/commands/cleaner_chore_switch.rb | 4 +-
.../main/ruby/shell/commands/clear_block_cache.rb | 1 +
.../main/ruby/shell/commands/clear_deadservers.rb | 12 +-
.../commands/disable_exceed_throttle_quota.rb | 2 +-
.../ruby/shell/commands/disable_rpc_throttle.rb | 2 +-
.../shell/commands/enable_exceed_throttle_quota.rb | 2 +-
.../ruby/shell/commands/enable_rpc_throttle.rb | 2 +-
.../src/main/ruby/shell/commands/is_disabled.rb | 6 +-
.../src/main/ruby/shell/commands/normalize.rb | 4 +-
.../main/ruby/shell/commands/normalizer_enabled.rb | 4 +-
.../main/ruby/shell/commands/normalizer_switch.rb | 4 +-
.../ruby/shell/commands/snapshot_cleanup_switch.rb | 2 +-
.../ruby/shell/commands/splitormerge_enabled.rb | 6 +-
.../ruby/shell/commands/splitormerge_switch.rb | 4 +-
hbase-shell/src/test/ruby/hbase/admin_test.rb | 188 ++++++++++++++-
hbase-shell/src/test/ruby/hbase/quotas_test.rb | 18 +-
src/main/asciidoc/_chapters/configuration.adoc | 1 +
src/site/site.xml | 8 +
src/site/xdoc/downloads.xml | 23 ++
81 files changed, 1742 insertions(+), 743 deletions(-)
create mode 100644 hbase-hadoop-compat/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableRangeHistogram.java
rename hbase-server/src/main/java/org/apache/hadoop/hbase/{regionserver/slowlog => namequeues}/DisruptorExceptionHandler.java (96%)
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/LogEventHandler.java
rename hbase-server/src/main/java/org/apache/hadoop/hbase/{regionserver/slowlog => namequeues}/LogHandlerUtils.java (96%)
copy hbase-common/src/main/java/org/apache/hadoop/hbase/util/ImmutableByteArray.java => hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/NamedQueuePayload.java (58%)
rename hbase-server/src/main/java/org/apache/hadoop/hbase/{regionserver/slowlog/SlowLogRecorder.java => namequeues/NamedQueueRecorder.java} (50%)
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/NamedQueueService.java
rename hbase-server/src/main/java/org/apache/hadoop/hbase/{regionserver/slowlog => namequeues}/RingBufferEnvelope.java (68%)
rename hbase-server/src/main/java/org/apache/hadoop/hbase/{regionserver/slowlog => namequeues}/RpcLogDetails.java (94%)
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/SlowLogPersistentService.java
rename hbase-server/src/main/java/org/apache/hadoop/hbase/{regionserver/slowlog => namequeues}/SlowLogTableOpsChore.java (84%)
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/impl/SlowLogQueueService.java
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/request/NamedQueueGetRequest.java
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/response/NamedQueueGetResponse.java
delete mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/slowlog/LogEventHandler.java
rename hbase-server/src/test/java/org/apache/hadoop/hbase/{regionserver/slowlog/TestSlowLogRecorder.java => namequeues/TestNamedQueueRecorder.java} (76%)
rename hbase-server/src/test/java/org/apache/hadoop/hbase/{regionserver/slowlog => namequeues}/TestSlowLogAccessor.java (76%)
[hbase] 01/02: HBASE-24682 Refactor ReplicationSource#addHFileRefs
method: move it to ReplicationSourceManager (#2020)
Posted by zg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zghao pushed a commit to branch HBASE-24666
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit bb9cae1b60b57106dff8fd6fd935ba98f01a0117
Author: Guanghao Zhang <zg...@apache.org>
AuthorDate: Wed Jul 8 14:29:08 2020 +0800
HBASE-24682 Refactor ReplicationSource#addHFileRefs method: move it to ReplicationSourceManager (#2020)
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
---
.../regionserver/ReplicationSource.java | 38 +----------------
.../regionserver/ReplicationSourceInterface.java | 14 -------
.../regionserver/ReplicationSourceManager.java | 48 +++++++++++++++++++++-
.../hbase/replication/ReplicationSourceDummy.java | 9 +---
4 files changed, 49 insertions(+), 60 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
index bc17549..5f8f2db 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
@@ -28,7 +28,6 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -36,6 +35,7 @@ import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -45,21 +45,17 @@ import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableDescriptors;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.RSRpcServices;
import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost;
import org.apache.hadoop.hbase.replication.ChainWALEntryFilter;
import org.apache.hadoop.hbase.replication.ClusterMarkingEntryFilter;
import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
-import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationPeer;
import org.apache.hadoop.hbase.replication.ReplicationQueueInfo;
import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
import org.apache.hadoop.hbase.replication.SystemTableWALEntryFilter;
import org.apache.hadoop.hbase.replication.WALEntryFilter;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.hadoop.hbase.wal.WAL.Entry;
@@ -224,38 +220,6 @@ public class ReplicationSource implements ReplicationSourceInterface {
}
}
- @Override
- public void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> pairs)
- throws ReplicationException {
- String peerId = replicationPeer.getId();
- Set<String> namespaces = replicationPeer.getNamespaces();
- Map<TableName, List<String>> tableCFMap = replicationPeer.getTableCFs();
- if (tableCFMap != null) { // All peers with TableCFs
- List<String> tableCfs = tableCFMap.get(tableName);
- if (tableCFMap.containsKey(tableName)
- && (tableCfs == null || tableCfs.contains(Bytes.toString(family)))) {
- this.queueStorage.addHFileRefs(peerId, pairs);
- metrics.incrSizeOfHFileRefsQueue(pairs.size());
- } else {
- LOG.debug("HFiles will not be replicated belonging to the table {} family {} to peer id {}",
- tableName, Bytes.toString(family), peerId);
- }
- } else if (namespaces != null) { // Only for set NAMESPACES peers
- if (namespaces.contains(tableName.getNamespaceAsString())) {
- this.queueStorage.addHFileRefs(peerId, pairs);
- metrics.incrSizeOfHFileRefsQueue(pairs.size());
- } else {
- LOG.debug("HFiles will not be replicated belonging to the table {} family {} to peer id {}",
- tableName, Bytes.toString(family), peerId);
- }
- } else {
- // user has explicitly not defined any table cfs for replication, means replicate all the
- // data
- this.queueStorage.addHFileRefs(peerId, pairs);
- metrics.incrSizeOfHFileRefsQueue(pairs.size());
- }
- }
-
private ReplicationEndpoint createReplicationEndpoint()
throws InstantiationException, IllegalAccessException, ClassNotFoundException, IOException {
RegionServerCoprocessorHost rsServerHost = null;
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
index 0bd90cf..33a413f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
@@ -29,12 +29,9 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.ServerName;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
-import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationPeer;
import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
-import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.yetus.audience.InterfaceAudience;
@@ -63,17 +60,6 @@ public interface ReplicationSourceInterface {
void enqueueLog(Path log);
/**
- * Add hfile names to the queue to be replicated.
- * @param tableName Name of the table these files belongs to
- * @param family Name of the family these files belong to
- * @param pairs list of pairs of { HFile location in staging dir, HFile path in region dir which
- * will be added in the queue for replication}
- * @throws ReplicationException If failed to add hfile references
- */
- void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> pairs)
- throws ReplicationException;
-
- /**
* Start the replication
*/
void startup();
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
index 1a012bd..a559b3d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
@@ -61,6 +61,7 @@ import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
import org.apache.hadoop.hbase.replication.ReplicationTracker;
import org.apache.hadoop.hbase.replication.ReplicationUtils;
import org.apache.hadoop.hbase.replication.SyncReplicationState;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.hadoop.hbase.wal.SyncReplicationWALProvider;
@@ -170,6 +171,8 @@ public class ReplicationSourceManager implements ReplicationListener {
// replication peer.
private final int maxRetriesMultiplier;
+ private final Map<String, MetricsSource> sourceMetrics = new HashMap<>();
+
/**
* Creates a replication manager and sets the watch on all the other registered region servers
* @param queueStorage the interface for manipulating replication queues
@@ -348,6 +351,7 @@ public class ReplicationSourceManager implements ReplicationListener {
ReplicationSourceInterface src = ReplicationSourceFactory.create(conf, queueId);
MetricsSource metrics = new MetricsSource(queueId);
+ sourceMetrics.put(queueId, metrics);
// init replication source
src.init(conf, fs, this, queueStorage, replicationPeer, server, queueId, clusterId,
walFileLengthProvider, metrics);
@@ -1120,7 +1124,49 @@ public class ReplicationSourceManager implements ReplicationListener {
public void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> pairs)
throws IOException {
for (ReplicationSourceInterface source : this.sources.values()) {
- throwIOExceptionWhenFail(() -> source.addHFileRefs(tableName, family, pairs));
+ throwIOExceptionWhenFail(() -> addHFileRefs(source.getPeerId(), tableName, family, pairs));
+ }
+ }
+
+ /**
+ * Add hfile names to the queue to be replicated.
+ * @param peerId the replication peer id
+ * @param tableName Name of the table these files belongs to
+ * @param family Name of the family these files belong to
+ * @param pairs list of pairs of { HFile location in staging dir, HFile path in region dir which
+ * will be added in the queue for replication}
+ * @throws ReplicationException If failed to add hfile references
+ */
+ private void addHFileRefs(String peerId, TableName tableName, byte[] family,
+ List<Pair<Path, Path>> pairs) throws ReplicationException {
+ // Only the normal replication source update here, its peerId is equals to queueId.
+ MetricsSource metrics = sourceMetrics.get(peerId);
+ ReplicationPeer replicationPeer = replicationPeers.getPeer(peerId);
+ Set<String> namespaces = replicationPeer.getNamespaces();
+ Map<TableName, List<String>> tableCFMap = replicationPeer.getTableCFs();
+ if (tableCFMap != null) { // All peers with TableCFs
+ List<String> tableCfs = tableCFMap.get(tableName);
+ if (tableCFMap.containsKey(tableName)
+ && (tableCfs == null || tableCfs.contains(Bytes.toString(family)))) {
+ this.queueStorage.addHFileRefs(peerId, pairs);
+ metrics.incrSizeOfHFileRefsQueue(pairs.size());
+ } else {
+ LOG.debug("HFiles will not be replicated belonging to the table {} family {} to peer id {}",
+ tableName, Bytes.toString(family), peerId);
+ }
+ } else if (namespaces != null) { // Only for set NAMESPACES peers
+ if (namespaces.contains(tableName.getNamespaceAsString())) {
+ this.queueStorage.addHFileRefs(peerId, pairs);
+ metrics.incrSizeOfHFileRefsQueue(pairs.size());
+ } else {
+ LOG.debug("HFiles will not be replicated belonging to the table {} family {} to peer id {}",
+ tableName, Bytes.toString(family), peerId);
+ }
+ } else {
+ // user has explicitly not defined any table cfs for replication, means replicate all the
+ // data
+ this.queueStorage.addHFileRefs(peerId, pairs);
+ metrics.incrSizeOfHFileRefsQueue(pairs.size());
}
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
index a361c44..781a1da 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
@@ -21,17 +21,16 @@ import java.io.IOException;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.ServerName;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.replication.regionserver.MetricsSource;
import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceInterface;
import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager;
import org.apache.hadoop.hbase.replication.regionserver.WALFileLengthProvider;
-import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.wal.WAL.Entry;
/**
@@ -114,12 +113,6 @@ public class ReplicationSourceDummy implements ReplicationSourceInterface {
}
@Override
- public void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> files)
- throws ReplicationException {
- return;
- }
-
- @Override
public boolean isPeerEnabled() {
return true;
}
[hbase] 02/02: HBASE-24681 Remove the cache
walsById/walsByIdRecoveredQueues from ReplicationSourceManager (#2019)
Posted by zg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zghao pushed a commit to branch HBASE-24666
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit fae9f0cd51816ef60c5a6a1fa9a9eaf263090b56
Author: Guanghao Zhang <zg...@apache.org>
AuthorDate: Mon Jul 13 17:35:32 2020 +0800
HBASE-24681 Remove the cache walsById/walsByIdRecoveredQueues from ReplicationSourceManager (#2019)
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
---
.../regionserver/ReplicationSourceManager.java | 204 +++++++--------------
1 file changed, 62 insertions(+), 142 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
index a559b3d..158a66e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
@@ -91,30 +91,6 @@ import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFacto
* <li>No need synchronized on {@link #sources}. {@link #sources} is a ConcurrentHashMap and there
* is a Lock for peer id in {@link PeerProcedureHandlerImpl}. So there is no race for peer
* operations.</li>
- * <li>Need synchronized on {@link #walsById}. There are four methods which modify it,
- * {@link #addPeer(String)}, {@link #removePeer(String)},
- * {@link #cleanOldLogs(String, boolean, ReplicationSourceInterface)} and {@link #preLogRoll(Path)}.
- * {@link #walsById} is a ConcurrentHashMap and there is a Lock for peer id in
- * {@link PeerProcedureHandlerImpl}. So there is no race between {@link #addPeer(String)} and
- * {@link #removePeer(String)}. {@link #cleanOldLogs(String, boolean, ReplicationSourceInterface)}
- * is called by {@link ReplicationSourceInterface}. So no race with {@link #addPeer(String)}.
- * {@link #removePeer(String)} will terminate the {@link ReplicationSourceInterface} firstly, then
- * remove the wals from {@link #walsById}. So no race with {@link #removePeer(String)}. The only
- * case need synchronized is {@link #cleanOldLogs(String, boolean, ReplicationSourceInterface)} and
- * {@link #preLogRoll(Path)}.</li>
- * <li>No need synchronized on {@link #walsByIdRecoveredQueues}. There are three methods which
- * modify it, {@link #removePeer(String)} ,
- * {@link #cleanOldLogs(String, boolean, ReplicationSourceInterface)} and
- * {@link ReplicationSourceManager.NodeFailoverWorker#run()}.
- * {@link #cleanOldLogs(String, boolean, ReplicationSourceInterface)} is called by
- * {@link ReplicationSourceInterface}. {@link #removePeer(String)} will terminate the
- * {@link ReplicationSourceInterface} firstly, then remove the wals from
- * {@link #walsByIdRecoveredQueues}. And {@link ReplicationSourceManager.NodeFailoverWorker#run()}
- * will add the wals to {@link #walsByIdRecoveredQueues} firstly, then start up a
- * {@link ReplicationSourceInterface}. So there is no race here. For
- * {@link ReplicationSourceManager.NodeFailoverWorker#run()} and {@link #removePeer(String)}, there
- * is already synchronized on {@link #oldsources}. So no need synchronized on
- * {@link #walsByIdRecoveredQueues}.</li>
* <li>Need synchronized on {@link #latestPaths} to avoid the new open source miss new log.</li>
* <li>Need synchronized on {@link #oldsources} to avoid adding recovered source for the
* to-be-removed peer.</li>
@@ -135,15 +111,6 @@ public class ReplicationSourceManager implements ReplicationListener {
// All about stopping
private final Server server;
- // All logs we are currently tracking
- // Index structure of the map is: queue_id->logPrefix/logGroup->logs
- // For normal replication source, the peer id is same with the queue id
- private final ConcurrentMap<String, Map<String, NavigableSet<String>>> walsById;
- // Logs for recovered sources we are currently tracking
- // the map is: queue_id->logPrefix/logGroup->logs
- // For recovered source, the queue id's format is peer_id-servername-*
- private final ConcurrentMap<String, Map<String, NavigableSet<String>>> walsByIdRecoveredQueues;
-
private final SyncReplicationPeerMappingManager syncReplicationPeerMappingManager;
private final Configuration conf;
@@ -195,8 +162,6 @@ public class ReplicationSourceManager implements ReplicationListener {
this.replicationPeers = replicationPeers;
this.replicationTracker = replicationTracker;
this.server = server;
- this.walsById = new ConcurrentHashMap<>();
- this.walsByIdRecoveredQueues = new ConcurrentHashMap<>();
this.oldsources = new ArrayList<>();
this.conf = conf;
this.fs = fs;
@@ -331,7 +296,6 @@ public class ReplicationSourceManager implements ReplicationListener {
// Delete queue from storage and memory and queue id is same with peer id for normal
// source
deleteQueue(peerId);
- this.walsById.remove(peerId);
}
ReplicationPeerConfig peerConfig = peer.getPeerConfig();
if (peerConfig.isSyncReplication()) {
@@ -372,15 +336,10 @@ public class ReplicationSourceManager implements ReplicationListener {
// synchronized on latestPaths to avoid missing the new log
synchronized (this.latestPaths) {
this.sources.put(peerId, src);
- Map<String, NavigableSet<String>> walsByGroup = new HashMap<>();
- this.walsById.put(peerId, walsByGroup);
// Add the latest wal to that source's queue
if (!latestPaths.isEmpty()) {
for (Map.Entry<String, Path> walPrefixAndPath : latestPaths.entrySet()) {
Path walPath = walPrefixAndPath.getValue();
- NavigableSet<String> wals = new TreeSet<>();
- wals.add(walPath.getName());
- walsByGroup.put(walPrefixAndPath.getKey(), wals);
// Abort RS and throw exception to make add peer failed
abortAndThrowIOExceptionWhenFail(
() -> this.queueStorage.addWAL(server.getServerName(), peerId, walPath.getName()));
@@ -434,7 +393,10 @@ public class ReplicationSourceManager implements ReplicationListener {
// map from walsById since later we may fail to delete them from the replication queue
// storage, and when we retry next time, we can not know the wal files that need to be deleted
// from the replication queue storage.
- walsById.get(peerId).forEach((k, v) -> wals.put(k, new TreeSet<>(v)));
+ this.queueStorage.getWALsInQueue(this.server.getServerName(), peerId).forEach(wal -> {
+ String walPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(wal);
+ wals.computeIfAbsent(walPrefix, p -> new TreeSet<>()).add(wal);
+ });
}
LOG.info("Startup replication source for " + src.getPeerId());
src.startup();
@@ -443,15 +405,6 @@ public class ReplicationSourceManager implements ReplicationListener {
queueStorage.removeWAL(server.getServerName(), peerId, wal);
}
}
- synchronized (walsById) {
- Map<String, NavigableSet<String>> oldWals = walsById.get(peerId);
- wals.forEach((k, v) -> {
- NavigableSet<String> walsByGroup = oldWals.get(k);
- if (walsByGroup != null) {
- walsByGroup.removeAll(v);
- }
- });
- }
// synchronized on oldsources to avoid race with NodeFailoverWorker. Since NodeFailoverWorker is
// a background task, we will delete the file from replication queue storage under the lock to
// simplify the logic.
@@ -463,7 +416,6 @@ public class ReplicationSourceManager implements ReplicationListener {
oldSource.terminate(terminateMessage);
oldSource.getSourceMetrics().clear();
queueStorage.removeQueue(server.getServerName(), queueId);
- walsByIdRecoveredQueues.remove(queueId);
iter.remove();
}
}
@@ -476,7 +428,7 @@ public class ReplicationSourceManager implements ReplicationListener {
* replication queue storage and only to enqueue all logs to the new replication source
* @param peerId the id of the replication peer
*/
- public void refreshSources(String peerId) throws IOException {
+ public void refreshSources(String peerId) throws ReplicationException, IOException {
String terminateMessage = "Peer " + peerId +
" state or config changed. Will close the previous replication source and open a new one";
ReplicationPeer peer = replicationPeers.getPeer(peerId);
@@ -489,9 +441,8 @@ public class ReplicationSourceManager implements ReplicationListener {
// Do not clear metrics
toRemove.terminate(terminateMessage, null, false);
}
- for (NavigableSet<String> walsByGroup : walsById.get(peerId).values()) {
- walsByGroup.forEach(wal -> src.enqueueLog(new Path(this.logDir, wal)));
- }
+ this.queueStorage.getWALsInQueue(this.server.getServerName(), peerId)
+ .forEach(wal -> src.enqueueLog(new Path(this.logDir, wal)));
}
LOG.info("Startup replication source for " + src.getPeerId());
src.startup();
@@ -512,9 +463,8 @@ public class ReplicationSourceManager implements ReplicationListener {
for (String queueId : previousQueueIds) {
ReplicationSourceInterface replicationSource = createSource(queueId, peer);
this.oldsources.add(replicationSource);
- for (SortedSet<String> walsByGroup : walsByIdRecoveredQueues.get(queueId).values()) {
- walsByGroup.forEach(wal -> src.enqueueLog(new Path(wal)));
- }
+ this.queueStorage.getWALsInQueue(this.server.getServerName(), queueId)
+ .forEach(wal -> src.enqueueLog(new Path(wal)));
toStartup.add(replicationSource);
}
}
@@ -534,7 +484,6 @@ public class ReplicationSourceManager implements ReplicationListener {
LOG.info("Done with the recovered queue {}", src.getQueueId());
// Delete queue from storage and memory
deleteQueue(src.getQueueId());
- this.walsByIdRecoveredQueues.remove(src.getQueueId());
return true;
}
@@ -557,8 +506,6 @@ public class ReplicationSourceManager implements ReplicationListener {
this.sources.remove(src.getPeerId());
// Delete queue from storage and memory
deleteQueue(src.getQueueId());
- this.walsById.remove(src.getQueueId());
-
}
/**
@@ -644,42 +591,19 @@ public class ReplicationSourceManager implements ReplicationListener {
* @param source the replication source
*/
@VisibleForTesting
- void cleanOldLogs(String log, boolean inclusive, ReplicationSourceInterface source) {
- String logPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(log);
- if (source.isRecovered()) {
- NavigableSet<String> wals = walsByIdRecoveredQueues.get(source.getQueueId()).get(logPrefix);
- if (wals != null) {
- NavigableSet<String> walsToRemove = wals.headSet(log, inclusive);
- if (walsToRemove.isEmpty()) {
- return;
- }
- cleanOldLogs(walsToRemove, source);
- walsToRemove.clear();
- }
- } else {
- NavigableSet<String> wals;
- NavigableSet<String> walsToRemove;
- // synchronized on walsById to avoid race with preLogRoll
- synchronized (this.walsById) {
- wals = walsById.get(source.getQueueId()).get(logPrefix);
- if (wals == null) {
- return;
- }
- walsToRemove = wals.headSet(log, inclusive);
- if (walsToRemove.isEmpty()) {
- return;
- }
- walsToRemove = new TreeSet<>(walsToRemove);
- }
- // cleanOldLogs may spend some time, especially for sync replication where we may want to
- // remove remote wals as the remote cluster may have already been down, so we do it outside
- // the lock to avoid block preLogRoll
- cleanOldLogs(walsToRemove, source);
- // now let's remove the files in the set
- synchronized (this.walsById) {
- wals.removeAll(walsToRemove);
- }
+ void cleanOldLogs(String log, boolean inclusive,
+ ReplicationSourceInterface source) {
+ NavigableSet<String> walsToRemove;
+ synchronized (this.latestPaths) {
+ walsToRemove = getWalsToRemove(source.getQueueId(), log, inclusive);
}
+ if (walsToRemove.isEmpty()) {
+ return;
+ }
+ // cleanOldLogs may spend some time, especially for sync replication where we may want to
+ // remove remote wals as the remote cluster may have already been down, so we do it outside
+ // the lock to avoid block preLogRoll
+ cleanOldLogs(walsToRemove, source);
}
private void removeRemoteWALs(String peerId, String remoteWALDir, Collection<String> wals)
@@ -760,37 +684,6 @@ public class ReplicationSourceManager implements ReplicationListener {
abortAndThrowIOExceptionWhenFail(
() -> this.queueStorage.addWAL(server.getServerName(), source.getQueueId(), logName));
}
-
- // synchronized on walsById to avoid race with cleanOldLogs
- synchronized (this.walsById) {
- // Update walsById map
- for (Map.Entry<String, Map<String, NavigableSet<String>>> entry : this.walsById
- .entrySet()) {
- String peerId = entry.getKey();
- Map<String, NavigableSet<String>> walsByPrefix = entry.getValue();
- boolean existingPrefix = false;
- for (Map.Entry<String, NavigableSet<String>> walsEntry : walsByPrefix.entrySet()) {
- SortedSet<String> wals = walsEntry.getValue();
- if (this.sources.isEmpty()) {
- // If there's no slaves, don't need to keep the old wals since
- // we only consider the last one when a new slave comes in
- wals.clear();
- }
- if (logPrefix.equals(walsEntry.getKey())) {
- wals.add(logName);
- existingPrefix = true;
- }
- }
- if (!existingPrefix) {
- // The new log belongs to a new group, add it into this peer
- LOG.debug("Start tracking logs for wal group {} for peer {}", logPrefix, peerId);
- NavigableSet<String> wals = new TreeSet<>();
- wals.add(logName);
- walsByPrefix.put(logPrefix, wals);
- }
- }
- }
-
// Add to latestPaths
latestPaths.put(logPrefix, newLog);
}
@@ -962,18 +855,6 @@ public class ReplicationSourceManager implements ReplicationListener {
continue;
}
}
- // track sources in walsByIdRecoveredQueues
- Map<String, NavigableSet<String>> walsByGroup = new HashMap<>();
- walsByIdRecoveredQueues.put(queueId, walsByGroup);
- for (String wal : walsSet) {
- String walPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(wal);
- NavigableSet<String> wals = walsByGroup.get(walPrefix);
- if (wals == null) {
- wals = new TreeSet<>();
- walsByGroup.put(walPrefix, wals);
- }
- wals.add(wal);
- }
oldsources.add(src);
LOG.trace("Added source for recovered queue: " + src.getQueueId());
for (String wal : walsSet) {
@@ -1005,7 +886,18 @@ public class ReplicationSourceManager implements ReplicationListener {
* @return a sorted set of wal names
*/
@VisibleForTesting
- public Map<String, Map<String, NavigableSet<String>>> getWALs() {
+ public Map<String, Map<String, NavigableSet<String>>> getWALs()
+ throws ReplicationException {
+ Map<String, Map<String, NavigableSet<String>>> walsById = new HashMap<>();
+ for (ReplicationSourceInterface source : sources.values()) {
+ String queueId = source.getQueueId();
+ Map<String, NavigableSet<String>> walsByGroup = new HashMap<>();
+ walsById.put(queueId, walsByGroup);
+ for (String wal : this.queueStorage.getWALsInQueue(this.server.getServerName(), queueId)) {
+ String walPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(wal);
+ walsByGroup.computeIfAbsent(walPrefix, p -> new TreeSet<>()).add(wal);
+ }
+ }
return Collections.unmodifiableMap(walsById);
}
@@ -1014,7 +906,18 @@ public class ReplicationSourceManager implements ReplicationListener {
* @return a sorted set of wal names
*/
@VisibleForTesting
- Map<String, Map<String, NavigableSet<String>>> getWalsByIdRecoveredQueues() {
+ Map<String, Map<String, NavigableSet<String>>> getWalsByIdRecoveredQueues()
+ throws ReplicationException {
+ Map<String, Map<String, NavigableSet<String>>> walsByIdRecoveredQueues = new HashMap<>();
+ for (ReplicationSourceInterface source : oldsources) {
+ String queueId = source.getQueueId();
+ Map<String, NavigableSet<String>> walsByGroup = new HashMap<>();
+ walsByIdRecoveredQueues.put(queueId, walsByGroup);
+ for (String wal : this.queueStorage.getWALsInQueue(this.server.getServerName(), queueId)) {
+ String walPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(wal);
+ walsByGroup.computeIfAbsent(walPrefix, p -> new TreeSet<>()).add(wal);
+ }
+ }
return Collections.unmodifiableMap(walsByIdRecoveredQueues);
}
@@ -1177,4 +1080,21 @@ public class ReplicationSourceManager implements ReplicationListener {
int activeFailoverTaskCount() {
return executor.getActiveCount();
}
+
+ private NavigableSet<String> getWalsToRemove(String queueId, String log, boolean inclusive) {
+ NavigableSet<String> walsToRemove = new TreeSet<>();
+ String logPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(log);
+ try {
+ this.queueStorage.getWALsInQueue(this.server.getServerName(), queueId).forEach(wal -> {
+ String walPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(wal);
+ if (walPrefix.equals(logPrefix)) {
+ walsToRemove.add(wal);
+ }
+ });
+ } catch (ReplicationException e) {
+ // Just log the exception here, as the recovered replication source will try to cleanup again.
+ LOG.warn("Failed to read wals in queue {}", queueId, e);
+ }
+ return walsToRemove.headSet(log, inclusive);
+ }
}