HBASE-26151 Reimplement MasterAddressTracker to also cache backup master addresses (#3548)

Signed-off-by: Yulin Niu <niuyulin@apache.org>
This commit is contained in:
Duo Zhang 2021-07-31 15:39:08 +08:00 committed by GitHub
parent 90f23d4743
commit b248730126
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 81 additions and 60 deletions

View File

@ -33,6 +33,7 @@ import static org.apache.hadoop.hbase.SplitLogCounters.tot_mgr_task_deleted;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
import java.io.IOException; import java.io.IOException;
import java.util.Map; import java.util.Map;
import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.atomic.LongAdder;
@ -51,10 +52,10 @@ import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination;
import org.apache.hadoop.hbase.coordination.ZkCoordinatedStateManager; import org.apache.hadoop.hbase.coordination.ZkCoordinatedStateManager;
import org.apache.hadoop.hbase.master.SplitLogManager.Task; import org.apache.hadoop.hbase.master.SplitLogManager.Task;
import org.apache.hadoop.hbase.master.SplitLogManager.TaskBatch; import org.apache.hadoop.hbase.master.SplitLogManager.TaskBatch;
import org.apache.hadoop.hbase.regionserver.TestMasterAddressTracker.NodeCreationListener;
import org.apache.hadoop.hbase.testclassification.LargeTests; import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.testclassification.MasterTests; import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.zookeeper.TestMasterAddressTracker.NodeCreationListener;
import org.apache.hadoop.hbase.zookeeper.ZKSplitLog; import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
import org.apache.hadoop.hbase.zookeeper.ZKUtil; import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZKWatcher; import org.apache.hadoop.hbase.zookeeper.ZKWatcher;

View File

@ -19,7 +19,6 @@ package org.apache.hadoop.hbase.zookeeper;
import java.io.IOException; import java.io.IOException;
import java.io.InterruptedIOException; import java.io.InterruptedIOException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.Comparator; import java.util.Comparator;
@ -33,6 +32,7 @@ import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.data.Stat;
import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException; import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos; import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos; import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos;
@ -57,6 +57,9 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos;
*/ */
@InterfaceAudience.Private @InterfaceAudience.Private
public class MasterAddressTracker extends ZKNodeTracker { public class MasterAddressTracker extends ZKNodeTracker {
private volatile List<ServerName> backupMasters = Collections.emptyList();
/** /**
* Construct a master address listener with the specified * Construct a master address listener with the specified
* <code>zookeeper</code> reference. * <code>zookeeper</code> reference.
@ -72,6 +75,26 @@ public class MasterAddressTracker extends ZKNodeTracker {
super(watcher, watcher.getZNodePaths().masterAddressZNode, abortable); super(watcher, watcher.getZNodePaths().masterAddressZNode, abortable);
} }
private void loadBackupMasters() {
try {
backupMasters = Collections.unmodifiableList(getBackupMastersAndRenewWatch(watcher));
} catch (InterruptedIOException e) {
abortable.abort("Unexpected exception handling nodeChildrenChanged event", e);
}
}
@Override
protected void postStart() {
loadBackupMasters();
}
@Override
public void nodeChildrenChanged(String path) {
if (path.equals(watcher.getZNodePaths().backupMasterAddressesZNode)) {
loadBackupMasters();
}
}
/** /**
* Get the address of the current master if one is available. Returns null * Get the address of the current master if one is available. Returns null
* if no current master. * if no current master.
@ -252,11 +275,12 @@ public class MasterAddressTracker extends ZKNodeTracker {
} }
int prefixLen = ProtobufUtil.lengthOfPBMagic(); int prefixLen = ProtobufUtil.lengthOfPBMagic();
try { try {
return ZooKeeperProtos.Master.PARSER.parseFrom(data, prefixLen, data.length - prefixLen); return ZooKeeperProtos.Master.parser().parseFrom(data, prefixLen, data.length - prefixLen);
} catch (InvalidProtocolBufferException e) { } catch (InvalidProtocolBufferException e) {
throw new DeserializationException(e); throw new DeserializationException(e);
} }
} }
/** /**
* delete the master znode if its content is same as the parameter * delete the master znode if its content is same as the parameter
* @param zkw must not be null * @param zkw must not be null
@ -284,7 +308,7 @@ public class MasterAddressTracker extends ZKNodeTracker {
} }
public List<ServerName> getBackupMasters() throws InterruptedIOException { public List<ServerName> getBackupMasters() throws InterruptedIOException {
return getBackupMastersAndRenewWatch(watcher); return backupMasters;
} }
/** /**

View File

@ -42,7 +42,7 @@ public abstract class ZKNodeTracker extends ZKListener {
protected final String node; protected final String node;
/** Data of the node being tracked */ /** Data of the node being tracked */
private byte [] data; private byte[] data;
/** Used to abort if a fatal error occurs */ /** Used to abort if a fatal error occurs */
protected final Abortable abortable; protected final Abortable abortable;
@ -51,16 +51,14 @@ public abstract class ZKNodeTracker extends ZKListener {
/** /**
* Constructs a new ZK node tracker. * Constructs a new ZK node tracker.
* * <p/>
* <p>After construction, use {@link #start} to kick off tracking. * After construction, use {@link #start} to kick off tracking.
*
* @param watcher reference to the {@link ZKWatcher} which also contains configuration and * @param watcher reference to the {@link ZKWatcher} which also contains configuration and
* constants * constants
* @param node path of the node being tracked * @param node path of the node being tracked
* @param abortable used to abort if a fatal error occurs * @param abortable used to abort if a fatal error occurs
*/ */
public ZKNodeTracker(ZKWatcher watcher, String node, public ZKNodeTracker(ZKWatcher watcher, String node, Abortable abortable) {
Abortable abortable) {
super(watcher); super(watcher);
this.node = node; this.node = node;
this.abortable = abortable; this.abortable = abortable;
@ -69,9 +67,9 @@ public abstract class ZKNodeTracker extends ZKListener {
/** /**
* Starts the tracking of the node in ZooKeeper. * Starts the tracking of the node in ZooKeeper.
* * <p/>
* <p>Use {@link #blockUntilAvailable()} to block until the node is available * Use {@link #blockUntilAvailable()} to block until the node is available or
* or {@link #getData(boolean)} to get the data of the node if it is available. * {@link #getData(boolean)} to get the data of the node if it is available.
*/ */
public synchronized void start() { public synchronized void start() {
this.watcher.registerListener(this); this.watcher.registerListener(this);
@ -89,6 +87,13 @@ public abstract class ZKNodeTracker extends ZKListener {
} catch (KeeperException e) { } catch (KeeperException e) {
abortable.abort("Unexpected exception during initialization, aborting", e); abortable.abort("Unexpected exception during initialization, aborting", e);
} }
postStart();
}
/**
* Called after start is called. Sub classes could implement this method to load more data on zk.
*/
protected void postStart() {
} }
public synchronized void stop() { public synchronized void stop() {

View File

@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.hadoop.hbase.regionserver; package org.apache.hadoop.hbase.zookeeper;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertFalse;
@ -25,16 +25,11 @@ import static org.junit.Assert.assertTrue;
import java.util.List; import java.util.List;
import java.util.concurrent.Semaphore; import java.util.concurrent.Semaphore;
import org.apache.hadoop.hbase.HBaseClassTestRule; import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtil; import org.apache.hadoop.hbase.HBaseZKTestingUtil;
import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.testclassification.MediumTests; import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests; import org.apache.hadoop.hbase.testclassification.ZKTests;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
import org.apache.hadoop.hbase.zookeeper.ZKListener;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
import org.junit.After; import org.junit.After;
import org.junit.AfterClass; import org.junit.AfterClass;
import org.junit.BeforeClass; import org.junit.BeforeClass;
@ -46,18 +41,19 @@ import org.junit.rules.TestName;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@Category({RegionServerTests.class, MediumTests.class}) @Category({ ZKTests.class, MediumTests.class })
public class TestMasterAddressTracker { public class TestMasterAddressTracker {
@ClassRule @ClassRule
public static final HBaseClassTestRule CLASS_RULE = public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestMasterAddressTracker.class); HBaseClassTestRule.forClass(TestMasterAddressTracker.class);
private static final Logger LOG = LoggerFactory.getLogger(TestMasterAddressTracker.class); private static final Logger LOG = LoggerFactory.getLogger(TestMasterAddressTracker.class);
private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); private final static HBaseZKTestingUtil TEST_UTIL = new HBaseZKTestingUtil();
// Cleaned up after each unit test. // Cleaned up after each unit test.
private static ZKWatcher zk; private ZKWatcher zk;
@Rule @Rule
public TestName name = new TestName(); public TestName name = new TestName();
@ -81,15 +77,15 @@ public class TestMasterAddressTracker {
@Test @Test
public void testDeleteIfEquals() throws Exception { public void testDeleteIfEquals() throws Exception {
final ServerName sn = ServerName.valueOf("localhost", 1234, final ServerName sn =
EnvironmentEdgeManager.currentTime()); ServerName.valueOf("localhost", 1234, EnvironmentEdgeManager.currentTime());
final MasterAddressTracker addressTracker = setupMasterTracker(sn, 1772); final MasterAddressTracker addressTracker = setupMasterTracker(sn, 1772);
try { try {
assertFalse("shouldn't have deleted wrong master server.", assertFalse("shouldn't have deleted wrong master server.",
MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), "some other string.")); MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), "some other string."));
} finally { } finally {
assertTrue("Couldn't clean up master", assertTrue("Couldn't clean up master",
MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), sn.toString())); MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), sn.toString()));
} }
} }
@ -99,9 +95,8 @@ public class TestMasterAddressTracker {
* @param infoPort if there is an active master, set its info port. * @param infoPort if there is an active master, set its info port.
*/ */
private MasterAddressTracker setupMasterTracker(final ServerName sn, final int infoPort) private MasterAddressTracker setupMasterTracker(final ServerName sn, final int infoPort)
throws Exception { throws Exception {
zk = new ZKWatcher(TEST_UTIL.getConfiguration(), zk = new ZKWatcher(TEST_UTIL.getConfiguration(), name.getMethodName(), null);
name.getMethodName(), null);
ZKUtil.createAndFailSilent(zk, zk.getZNodePaths().baseZNode); ZKUtil.createAndFailSilent(zk, zk.getZNodePaths().baseZNode);
ZKUtil.createAndFailSilent(zk, zk.getZNodePaths().backupMasterAddressesZNode); ZKUtil.createAndFailSilent(zk, zk.getZNodePaths().backupMasterAddressesZNode);
@ -112,14 +107,14 @@ public class TestMasterAddressTracker {
zk.registerListener(addressTracker); zk.registerListener(addressTracker);
// Use a listener to capture when the node is actually created // Use a listener to capture when the node is actually created
NodeCreationListener listener = new NodeCreationListener(zk, NodeCreationListener listener =
zk.getZNodePaths().masterAddressZNode); new NodeCreationListener(zk, zk.getZNodePaths().masterAddressZNode);
zk.registerListener(listener); zk.registerListener(listener);
if (sn != null) { if (sn != null) {
LOG.info("Creating master node"); LOG.info("Creating master node");
MasterAddressTracker.setMasterAddress(zk, zk.getZNodePaths().masterAddressZNode, MasterAddressTracker.setMasterAddress(zk, zk.getZNodePaths().masterAddressZNode, sn,
sn, infoPort); infoPort);
// Wait for the node to be created // Wait for the node to be created
LOG.info("Waiting for master address manager to be notified"); LOG.info("Waiting for master address manager to be notified");
@ -130,16 +125,15 @@ public class TestMasterAddressTracker {
} }
/** /**
* Unit tests that uses ZooKeeper but does not use the master-side methods * Unit tests that uses ZooKeeper but does not use the master-side methods but rather acts
* but rather acts directly on ZK. * directly on ZK.
* @throws Exception
*/ */
@Test @Test
public void testMasterAddressTrackerFromZK() throws Exception { public void testMasterAddressTrackerFromZK() throws Exception {
// Create the master node with a dummy address // Create the master node with a dummy address
final int infoPort = 1235; final int infoPort = 1235;
final ServerName sn = ServerName.valueOf("localhost", 1234, final ServerName sn =
EnvironmentEdgeManager.currentTime()); ServerName.valueOf("localhost", 1234, EnvironmentEdgeManager.currentTime());
final MasterAddressTracker addressTracker = setupMasterTracker(sn, infoPort); final MasterAddressTracker addressTracker = setupMasterTracker(sn, infoPort);
try { try {
assertTrue(addressTracker.hasMaster()); assertTrue(addressTracker.hasMaster());
@ -148,11 +142,10 @@ public class TestMasterAddressTracker {
assertEquals(infoPort, addressTracker.getMasterInfoPort()); assertEquals(infoPort, addressTracker.getMasterInfoPort());
} finally { } finally {
assertTrue("Couldn't clean up master", assertTrue("Couldn't clean up master",
MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), sn.toString())); MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), sn.toString()));
} }
} }
@Test @Test
public void testParsingNull() throws Exception { public void testParsingNull() throws Exception {
assertNull("parse on null data should return null.", MasterAddressTracker.parse(null)); assertNull("parse on null data should return null.", MasterAddressTracker.parse(null));
@ -160,17 +153,16 @@ public class TestMasterAddressTracker {
@Test @Test
public void testNoBackups() throws Exception { public void testNoBackups() throws Exception {
final ServerName sn = ServerName.valueOf("localhost", 1234, final ServerName sn =
EnvironmentEdgeManager.currentTime()); ServerName.valueOf("localhost", 1234, EnvironmentEdgeManager.currentTime());
final MasterAddressTracker addressTracker = setupMasterTracker(sn, 1772); final MasterAddressTracker addressTracker = setupMasterTracker(sn, 1772);
try { try {
assertEquals("Should receive 0 for backup not found.", 0, assertEquals("Should receive 0 for backup not found.", 0,
addressTracker.getBackupMasterInfoPort( addressTracker.getBackupMasterInfoPort(ServerName.valueOf("doesnotexist.example.com", 1234,
ServerName.valueOf("doesnotexist.example.com", 1234, EnvironmentEdgeManager.currentTime())));
EnvironmentEdgeManager.currentTime())));
} finally { } finally {
assertTrue("Couldn't clean up master", assertTrue("Couldn't clean up master",
MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), sn.toString())); MasterAddressTracker.deleteIfEquals(addressTracker.getWatcher(), sn.toString()));
} }
} }
@ -184,8 +176,8 @@ public class TestMasterAddressTracker {
@Test @Test
public void testBackupMasters() throws Exception { public void testBackupMasters() throws Exception {
final ServerName sn = ServerName.valueOf("localhost", 5678, final ServerName sn =
EnvironmentEdgeManager.currentTime()); ServerName.valueOf("localhost", 5678, EnvironmentEdgeManager.currentTime());
final MasterAddressTracker addressTracker = setupMasterTracker(sn, 1111); final MasterAddressTracker addressTracker = setupMasterTracker(sn, 1111);
assertTrue(addressTracker.hasMaster()); assertTrue(addressTracker.hasMaster());
ServerName activeMaster = addressTracker.getMasterAddress(); ServerName activeMaster = addressTracker.getMasterAddress();
@ -195,13 +187,14 @@ public class TestMasterAddressTracker {
assertEquals(0, backupMasters.size()); assertEquals(0, backupMasters.size());
ServerName backupMaster1 = ServerName.valueOf("localhost", 2222, -1); ServerName backupMaster1 = ServerName.valueOf("localhost", 2222, -1);
ServerName backupMaster2 = ServerName.valueOf("localhost", 3333, -1); ServerName backupMaster2 = ServerName.valueOf("localhost", 3333, -1);
String backupZNode1 = ZNodePaths.joinZNode( String backupZNode1 =
zk.getZNodePaths().backupMasterAddressesZNode, backupMaster1.toString()); ZNodePaths.joinZNode(zk.getZNodePaths().backupMasterAddressesZNode, backupMaster1.toString());
String backupZNode2 = ZNodePaths.joinZNode( String backupZNode2 =
zk.getZNodePaths().backupMasterAddressesZNode, backupMaster2.toString()); ZNodePaths.joinZNode(zk.getZNodePaths().backupMasterAddressesZNode, backupMaster2.toString());
// Add a backup master // Add backup masters
MasterAddressTracker.setMasterAddress(zk, backupZNode1, backupMaster1, 2222); MasterAddressTracker.setMasterAddress(zk, backupZNode1, backupMaster1, 2222);
MasterAddressTracker.setMasterAddress(zk, backupZNode2, backupMaster2, 3333); MasterAddressTracker.setMasterAddress(zk, backupZNode2, backupMaster2, 3333);
TEST_UTIL.waitFor(30000, () -> addressTracker.getBackupMasters().size() == 2);
backupMasters = addressTracker.getBackupMasters(); backupMasters = addressTracker.getBackupMasters();
assertEquals(2, backupMasters.size()); assertEquals(2, backupMasters.size());
assertTrue(backupMasters.contains(backupMaster1)); assertTrue(backupMasters.contains(backupMaster1));
@ -222,7 +215,7 @@ public class TestMasterAddressTracker {
@Override @Override
public void nodeCreated(String path) { public void nodeCreated(String path) {
if(path.equals(node)) { if (path.equals(node)) {
LOG.debug("nodeCreated(" + path + ")"); LOG.debug("nodeCreated(" + path + ")");
lock.release(); lock.release();
} }
@ -232,6 +225,4 @@ public class TestMasterAddressTracker {
lock.acquire(); lock.acquire();
} }
} }
} }