Skip to content

Commit

Permalink
HBASE-27806 Support dynamic reinitializing replication peer storage (#…
Browse files Browse the repository at this point in the history
…5195)

Signed-off-by: Liangjun He <[email protected]>
(cherry picked from commit 18ae733)
  • Loading branch information
Apache9 committed Apr 23, 2023
1 parent e9ffc1b commit fe2992a
Show file tree
Hide file tree
Showing 7 changed files with 205 additions and 22 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,13 @@
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.conf.ConfigurationObserver;
import org.apache.yetus.audience.InterfaceAudience;

@InterfaceAudience.Private
public class ReplicationPeerImpl implements ReplicationPeer {
public class ReplicationPeerImpl implements ReplicationPeer, ConfigurationObserver {

private final Configuration conf;
private volatile Configuration conf;

private final String id;

Expand Down Expand Up @@ -122,4 +123,9 @@ public long getPeerBandwidth() {
public void registerPeerConfigListener(ReplicationPeerConfigListener listener) {
this.peerConfigListeners.add(listener);
}

@Override
public void onConfigurationChange(Configuration conf) {
this.conf = conf;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -24,25 +24,38 @@
import java.util.concurrent.ConcurrentMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hbase.conf.ConfigurationObserver;
import org.apache.hadoop.hbase.replication.ReplicationPeer.PeerState;
import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* This provides an class for maintaining a set of peer clusters. These peers are remote slave
* clusters that data is replicated to.
* <p>
* We implement {@link ConfigurationObserver} mainly for recreating the
* {@link ReplicationPeerStorage}, so we can change the {@link ReplicationPeerStorage} without
* restarting the region server.
*/
@InterfaceAudience.Private
public class ReplicationPeers {
public class ReplicationPeers implements ConfigurationObserver {

private final Configuration conf;
private static final Logger LOG = LoggerFactory.getLogger(ReplicationPeers.class);

private volatile Configuration conf;

// Map of peer clusters keyed by their id
private final ConcurrentMap<String, ReplicationPeerImpl> peerCache;
private final ReplicationPeerStorage peerStorage;
private final FileSystem fs;
private final ZKWatcher zookeeper;
private volatile ReplicationPeerStorage peerStorage;

ReplicationPeers(FileSystem fs, ZKWatcher zookeeper, Configuration conf) {
this.conf = conf;
this.fs = fs;
this.zookeeper = zookeeper;
this.peerCache = new ConcurrentHashMap<>();
this.peerStorage = ReplicationStorageFactory.getReplicationPeerStorage(fs, zookeeper, conf);
}
Expand Down Expand Up @@ -134,4 +147,18 @@ private ReplicationPeerImpl createPeer(String peerId) throws ReplicationExceptio
return new ReplicationPeerImpl(ReplicationUtils.getPeerClusterConfiguration(peerConfig, conf),
peerId, enabled, peerConfig);
}

@Override
public void onConfigurationChange(Configuration conf) {
this.conf = conf;
this.peerStorage = ReplicationStorageFactory.getReplicationPeerStorage(fs, zookeeper, conf);
for (ReplicationPeerImpl peer : peerCache.values()) {
try {
peer.onConfigurationChange(
ReplicationUtils.getPeerClusterConfiguration(peer.getPeerConfig(), conf));
} catch (ReplicationException e) {
LOG.warn("failed to reload configuration for peer {}", peer.getId(), e);
}
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,6 @@
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.client.TableState;
import org.apache.hadoop.hbase.conf.ConfigurationManager;
import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.exceptions.MasterStoppedException;
Expand Down Expand Up @@ -770,6 +769,7 @@ private void initializeZKBasedSystemTrackers()

this.replicationPeerManager =
ReplicationPeerManager.create(fileSystemManager.getFileSystem(), zooKeeper, conf, clusterId);
this.configurationManager.registerObserver(replicationPeerManager);
this.replicationPeerModificationStateStore =
new ReplicationPeerModificationStateStore(masterRegion);

Expand Down Expand Up @@ -4235,12 +4235,6 @@ public void onConfigurationChange(Configuration newConf) {
}
}

@RestrictedApi(explanation = "Should only be called in tests", link = "",
allowedOnPath = ".*/src/test/.*")
public ConfigurationManager getConfigurationManager() {
return configurationManager;
}

private void setQuotasObserver(Configuration conf) {
// Add the Observer to delete quotas on table deletion before starting all CPs by
// default with quota support, avoiding if user specifically asks to not load this Observer.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.replication.ReplicationPeerConfigUtil;
import org.apache.hadoop.hbase.conf.ConfigurationObserver;
import org.apache.hadoop.hbase.replication.BaseReplicationEndpoint;
import org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint;
import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
Expand All @@ -59,22 +60,33 @@
* Manages and performs all replication admin operations.
* <p>
* Used to add/remove a replication peer.
* <p>
* Implement {@link ConfigurationObserver} mainly for recreating {@link ReplicationPeerStorage}, for
* supporting migrating across different replication peer storages without restarting master.
*/
@InterfaceAudience.Private
public class ReplicationPeerManager {
public class ReplicationPeerManager implements ConfigurationObserver {

private final ReplicationPeerStorage peerStorage;
private volatile ReplicationPeerStorage peerStorage;

private final ReplicationQueueStorage queueStorage;

private final ConcurrentMap<String, ReplicationPeerDescription> peers;

private final String clusterId;

private final Configuration conf;
private volatile Configuration conf;

// for dynamic recreating ReplicationPeerStorage.
private final FileSystem fs;

private final ZKWatcher zk;

ReplicationPeerManager(ReplicationPeerStorage peerStorage, ReplicationQueueStorage queueStorage,
ConcurrentMap<String, ReplicationPeerDescription> peers, Configuration conf, String clusterId) {
ReplicationPeerManager(FileSystem fs, ZKWatcher zk, ReplicationPeerStorage peerStorage,
ReplicationQueueStorage queueStorage, ConcurrentMap<String, ReplicationPeerDescription> peers,
Configuration conf, String clusterId) {
this.fs = fs;
this.zk = zk;
this.peerStorage = peerStorage;
this.queueStorage = queueStorage;
this.peers = peers;
Expand Down Expand Up @@ -426,7 +438,7 @@ public static ReplicationPeerManager create(FileSystem fs, ZKWatcher zk, Configu
boolean enabled = peerStorage.isPeerEnabled(peerId);
peers.put(peerId, new ReplicationPeerDescription(peerId, enabled, peerConfig));
}
return new ReplicationPeerManager(peerStorage,
return new ReplicationPeerManager(fs, zk, peerStorage,
ReplicationStorageFactory.getReplicationQueueStorage(zk, conf), peers, conf, clusterId);
}

Expand All @@ -440,4 +452,10 @@ private boolean isStringEquals(String s1, String s2) {
}
return s1.equals(s2);
}

@Override
public void onConfigurationChange(Configuration conf) {
this.conf = conf;
this.peerStorage = ReplicationStorageFactory.getReplicationPeerStorage(fs, zk, conf);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import static org.apache.hadoop.hbase.replication.regionserver.ReplicationMarkerChore.REPLICATION_MARKER_ENABLED_KEY;
import static org.apache.hadoop.hbase.util.DNS.UNSAFE_RS_HOSTNAME_KEY;

import com.google.errorprone.annotations.RestrictedApi;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
Expand Down Expand Up @@ -2366,6 +2367,17 @@ private void initializeThreads() {
}

private void registerConfigurationObservers() {
// Register Replication if possible, as now we support recreating replication peer storage, for
// migrating across different replication peer storages online
if (replicationSourceHandler instanceof ConfigurationObserver) {
configurationManager.registerObserver((ConfigurationObserver) replicationSourceHandler);
}
if (
replicationSourceHandler != replicationSinkHandler
&& replicationSinkHandler instanceof ConfigurationObserver
) {
configurationManager.registerObserver((ConfigurationObserver) replicationSinkHandler);
}
// Registering the compactSplitThread object with the ConfigurationManager.
configurationManager.registerObserver(this.compactSplitThread);
configurationManager.registerObserver(this.rpcServices);
Expand Down Expand Up @@ -3821,8 +3833,9 @@ public ZKPermissionWatcher getZKPermissionWatcher() {
}

/** Returns : Returns the ConfigurationManager object for testing purposes. */
@InterfaceAudience.Private
ConfigurationManager getConfigurationManager() {
@RestrictedApi(explanation = "Should only be called in tests", link = "",
allowedOnPath = ".*/src/test/.*")
public ConfigurationManager getConfigurationManager() {
return configurationManager;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.conf.ConfigurationManager;
import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost;
import org.apache.hadoop.hbase.regionserver.ReplicationSinkService;
Expand All @@ -56,15 +58,19 @@

/**
* Gateway to Replication. Used by {@link org.apache.hadoop.hbase.regionserver.HRegionServer}.
* <p>
* Implement {@link PropagatingConfigurationObserver} mainly for registering
* {@link ReplicationPeers}, so we can recreating the replication peer storage.
*/
@InterfaceAudience.Private
public class Replication implements ReplicationSourceService, ReplicationSinkService {
public class Replication
implements ReplicationSourceService, ReplicationSinkService, PropagatingConfigurationObserver {
private static final Logger LOG = LoggerFactory.getLogger(Replication.class);
private boolean isReplicationForBulkLoadDataEnabled;
private ReplicationSourceManager replicationManager;
private ReplicationQueueStorage queueStorage;
private ReplicationPeers replicationPeers;
private Configuration conf;
private volatile Configuration conf;
private ReplicationSink replicationSink;
// Hosting server
private Server server;
Expand Down Expand Up @@ -262,4 +268,19 @@ private void buildReplicationLoad() {
MetricsSink sinkMetrics = this.replicationSink.getSinkMetrics();
this.replicationLoad.buildReplicationLoad(allSources, sinkMetrics);
}

@Override
public void onConfigurationChange(Configuration conf) {
this.conf = conf;
}

@Override
public void registerChildren(ConfigurationManager manager) {
manager.registerObserver(replicationPeers);
}

@Override
public void deregisterChildren(ConfigurationManager manager) {
manager.deregisterObserver(replicationPeers);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hbase.replication;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.testclassification.ReplicationTests;
import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread;
import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
import org.apache.hadoop.util.ToolRunner;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

@Category({ ReplicationTests.class, LargeTests.class })
public class TestMigrateRepliationPeerStorageOnline {

@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestMigrateRepliationPeerStorageOnline.class);

private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();

@BeforeClass
public static void setUp() throws Exception {
// use zookeeper first, and then migrate to filesystem
UTIL.getConfiguration().set(ReplicationStorageFactory.REPLICATION_PEER_STORAGE_IMPL,
ReplicationPeerStorageType.ZOOKEEPER.name());
UTIL.startMiniCluster(1);
}

@AfterClass
public static void tearDown() throws IOException {
UTIL.shutdownMiniCluster();
}

@Test
public void testMigrate() throws Exception {
Admin admin = UTIL.getAdmin();
ReplicationPeerConfig rpc =
ReplicationPeerConfig.newBuilder().setClusterKey(UTIL.getClusterKey() + "-test")
.setReplicationEndpointImpl(DummyReplicationEndpoint.class.getName()).build();
admin.addReplicationPeer("1", rpc);

// disable peer modification
admin.replicationPeerModificationSwitch(false, true);

// migrate replication peer data
Configuration conf = new Configuration(UTIL.getConfiguration());
assertEquals(0, ToolRunner.run(conf, new CopyReplicationPeers(conf),
new String[] { "zookeeper", "filesystem" }));
conf.set(ReplicationStorageFactory.REPLICATION_PEER_STORAGE_IMPL,
ReplicationPeerStorageType.FILESYSTEM.name());
// confirm that we have copied the data
ReplicationPeerStorage fsPeerStorage = ReplicationStorageFactory
.getReplicationPeerStorage(UTIL.getTestFileSystem(), UTIL.getZooKeeperWatcher(), conf);
assertNotNull(fsPeerStorage.getPeerConfig("1"));

for (MasterThread mt : UTIL.getMiniHBaseCluster().getMasterThreads()) {
Configuration newConf = new Configuration(mt.getMaster().getConfiguration());
newConf.set(ReplicationStorageFactory.REPLICATION_PEER_STORAGE_IMPL,
ReplicationPeerStorageType.FILESYSTEM.name());
mt.getMaster().getConfigurationManager().notifyAllObservers(newConf);
}
for (RegionServerThread rt : UTIL.getMiniHBaseCluster().getRegionServerThreads()) {
Configuration newConf = new Configuration(rt.getRegionServer().getConfiguration());
newConf.set(ReplicationStorageFactory.REPLICATION_PEER_STORAGE_IMPL,
ReplicationPeerStorageType.FILESYSTEM.name());
rt.getRegionServer().getConfigurationManager().notifyAllObservers(newConf);
}

admin.replicationPeerModificationSwitch(true);
admin.removeReplicationPeer("1");

// confirm that we will operation on the new peer storage
assertThat(fsPeerStorage.listPeerIds(), empty());
}
}

0 comments on commit fe2992a

Please sign in to comment.