You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by iv...@apache.org on 2012/08/27 17:44:00 UTC
svn commit: r1377716 - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/
bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/
Author: ivank
Date: Mon Aug 27 15:43:59 2012
New Revision: 1377716
URL: http://svn.apache.org/viewvc?rev=1377716&view=rev
Log:
BOOKKEEPER-272: Provide automatic mechanism to know bookie failures (rakeshr via ivank)
Added:
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/AuditorElector.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorBookieTest.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1377716&r1=1377715&r2=1377716&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Mon Aug 27 15:43:59 2012
@@ -104,6 +104,8 @@ Trunk (unreleased changes)
BOOKKEEPER-304: Prepare bookie vs ledgers cache and will be used by the Auditor (rakeshr via ivank)
+ BOOKKEEPER-272: Provide automatic mechanism to know bookie failures (rakeshr via ivank)
+
hedwig-server:
BOOKKEEPER-250: Need a ledger manager like interface to manage metadata operations in Hedwig (sijie via ivank)
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java?rev=1377716&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java Mon Aug 27 15:43:59 2012
@@ -0,0 +1,234 @@
+/**
+ *
+ * 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.bookkeeper.replication;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
+import org.apache.bookkeeper.replication.ReplicationException.BKAuditException;
+import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
+import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Auditor is a single entity in the entire Bookie cluster and will be watching
+ * all the bookies under 'ledgerrootpath/available' zkpath. When any of the
+ * bookie failed or disconnected from zk, he will start initiating the
+ * re-replication activities by keeping all the corresponding ledgers of the
+ * failed bookie as underreplicated znode in zk.
+ */
+public class Auditor extends Thread implements Watcher {
+ private static final Logger LOG = LoggerFactory.getLogger(Auditor.class);
+ private final LinkedBlockingQueue<EventType> bookieNotifications = new LinkedBlockingQueue<EventType>();
+ private final AbstractConfiguration conf;
+ private final ZooKeeper zkc;
+ private BookieLedgerIndexer bookieLedgerIndexer;
+ private LedgerUnderreplicationManager ledgerUnderreplicationManager;
+ private volatile boolean running = true;
+
+ public Auditor(String bookieIdentifier, AbstractConfiguration conf,
+ ZooKeeper zkc) throws UnavailableException {
+ setName("AuditorBookie-" + bookieIdentifier);
+ setDaemon(true);
+ this.conf = conf;
+ this.zkc = zkc;
+ initialize(conf, zkc);
+ }
+
+ private void initialize(AbstractConfiguration conf, ZooKeeper zkc)
+ throws UnavailableException {
+ try {
+ LedgerManagerFactory ledgerManagerFactory = LedgerManagerFactory
+ .newLedgerManagerFactory(conf, zkc);
+
+ this.bookieLedgerIndexer = new BookieLedgerIndexer(
+ ledgerManagerFactory.newLedgerManager());
+
+ this.ledgerUnderreplicationManager = ledgerManagerFactory
+ .newLedgerUnderreplicationManager();
+
+ } catch (CompatibilityException ce) {
+ throw new UnavailableException(
+ "CompatibilityException while initializing Auditor", ce);
+ } catch (IOException ioe) {
+ throw new UnavailableException(
+ "IOException while initializing Auditor", ioe);
+ } catch (KeeperException ke) {
+ throw new UnavailableException(
+ "KeeperException while initializing Auditor", ke);
+ } catch (InterruptedException ie) {
+ throw new UnavailableException(
+ "Interrupted while initializing Auditor", ie);
+ }
+ }
+
+ @Override
+ public void run() {
+ LOG.info("I'm starting as Auditor Bookie");
+ try {
+ // on startup watching available bookie and based on the
+ // available bookies determining the bookie failures.
+ List<String> knownBookies = getAvailableBookies();
+ auditingBookies(knownBookies);
+
+ while (true) {
+ // wait for bookie join/failure notifications
+ bookieNotifications.take();
+ List<String> availableBookies = getAvailableBookies();
+
+ // casting to String, as knownBookies and availableBookies
+ // contains only String values
+ // find new bookies(if any) and update the known bookie list
+ Collection<String> newBookies = CollectionUtils.subtract(
+ availableBookies, knownBookies);
+ knownBookies.addAll(newBookies);
+
+ // find lost bookies(if any)
+ Collection<String> lostBookies = CollectionUtils.subtract(
+ knownBookies, availableBookies);
+
+ if (lostBookies.size() > 0) {
+ knownBookies.removeAll(lostBookies);
+ Map<String, Set<Long>> ledgerDetails = generateBookie2LedgersIndex();
+ handleLostBookies(lostBookies, ledgerDetails);
+ }
+ }
+ } catch (KeeperException ke) {
+ LOG.error("Exception while watching available bookies", ke);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ LOG.error("Interrupted while watching available bookies ", ie);
+ } catch (BKAuditException bke) {
+ LOG.error("Exception while watching available bookies", bke);
+ }
+
+ shutdown();
+ }
+
+ private List<String> getAvailableBookies() throws KeeperException,
+ InterruptedException {
+ return zkc.getChildren(conf.getZkAvailableBookiesPath(), this);
+ }
+
+ private void auditingBookies(List<String> availableBookies)
+ throws BKAuditException, KeeperException, InterruptedException {
+
+ Map<String, Set<Long>> ledgerDetails = generateBookie2LedgersIndex();
+
+ // find lost bookies
+ Set<String> knownBookies = ledgerDetails.keySet();
+ Collection<String> lostBookies = CollectionUtils.subtract(knownBookies,
+ availableBookies);
+
+ if (lostBookies.size() > 0)
+ handleLostBookies(lostBookies, ledgerDetails);
+ }
+
+ private Map<String, Set<Long>> generateBookie2LedgersIndex()
+ throws BKAuditException {
+ return bookieLedgerIndexer.getBookieToLedgerIndex();
+ }
+
+ private void handleLostBookies(Collection<String> lostBookies,
+ Map<String, Set<Long>> ledgerDetails) throws BKAuditException,
+ KeeperException, InterruptedException {
+ LOG.info("Following are the failed bookies: " + lostBookies
+ + " and searching its ledgers for re-replication");
+
+ for (String bookieIP : lostBookies) {
+ // identify all the ledgers in bookieIP and publishing these ledgers
+ // as under-replicated.
+ publishSuspectedLedgers(bookieIP, ledgerDetails.get(bookieIP));
+ }
+ }
+
+ private void publishSuspectedLedgers(String bookieIP, Set<Long> ledgers)
+ throws KeeperException, InterruptedException, BKAuditException {
+ if (null == ledgers || ledgers.size() == 0) {
+ // there is no ledgers available for this bookie and just
+ // ignoring the bookie failures
+ LOG.info("There is no ledgers for the failed bookie: " + bookieIP);
+ return;
+ }
+ LOG.info("Following ledgers: " + ledgers + " of bookie: " + bookieIP
+ + " are identified as underreplicated");
+ for (Long ledgerId : ledgers) {
+ try {
+ ledgerUnderreplicationManager.markLedgerUnderreplicated(
+ ledgerId, bookieIP);
+ } catch (UnavailableException ue) {
+ throw new BKAuditException(
+ "Failed to publish underreplicated ledger: " + ledgerId
+ + " of bookie: " + bookieIP, ue);
+ }
+ }
+ }
+
+ @Override
+ public void process(WatchedEvent event) {
+ // listen children changed event from ZooKeeper
+ if (event.getType() == EventType.NodeChildrenChanged) {
+ if (running)
+ bookieNotifications.add(event.getType());
+ }
+ }
+
+ /**
+ * Shutdown the auditor
+ */
+ public void shutdown() {
+ if (!running) {
+ return;
+ }
+ running = false;
+ LOG.info("Shutting down " + getName());
+ this.interrupt();
+ try {
+ this.join();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ LOG.warn("Interrupted while shutting down auditor bookie", ie);
+ }
+ }
+
+ /**
+ * Return true if auditor is running otherwise return false
+ *
+ * @return auditor status
+ */
+ public boolean isRunning() {
+ return running;
+ }
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/AuditorElector.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/AuditorElector.java?rev=1377716&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/AuditorElector.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/AuditorElector.java Mon Aug 27 15:43:59 2012
@@ -0,0 +1,269 @@
+/**
+ *
+ * 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.bookkeeper.replication;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.io.Serializable;
+
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.commons.lang.StringUtils;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Performing auditor election using Apache ZooKeeper. Using ZooKeeper as a
+ * coordination service, when a bookie bids for auditor, it creates an ephemeral
+ * sequential file (znode) on ZooKeeper and considered as their vote. Vote
+ * format is 'V_sequencenumber'. Election will be done by comparing the
+ * ephemeral sequential numbers and the bookie which has created the least znode
+ * will be elected as Auditor. All the other bookies will be watching on their
+ * predecessor znode according to the ephemeral sequence numbers.
+ */
+public class AuditorElector {
+ private static final Logger LOG = LoggerFactory
+ .getLogger(AuditorElector.class);
+ // Represents the index of the auditor node
+ private static final int AUDITOR_INDEX = 0;
+ // Represents vote prefix
+ private static final String VOTE_PREFIX = "V_";
+ // Represents path Separator
+ private static final String PATH_SEPARATOR = "/";
+ // Represents urLedger path in zk
+ private final String basePath;
+ // Represents auditor election path in zk
+ private final String electionPath;
+
+ private final String bookieId;
+ private final AbstractConfiguration conf;
+ private final ZooKeeper zkc;
+
+ private String myVote;
+ private Auditor auditor;
+ private volatile boolean running = true;
+
+ /**
+ * AuditorElector for performing the auditor election
+ *
+ * @param bookieId
+ * - bookie identifier, comprises HostAddress:Port
+ * @param conf
+ * - configuration
+ * @param zkc
+ * - ZK instance
+ * @throws UnavailableException
+ * throws unavailable exception while initializing the elector
+ */
+ public AuditorElector(String bookieId, AbstractConfiguration conf,
+ ZooKeeper zkc) throws UnavailableException {
+ this.bookieId = bookieId;
+ this.conf = conf;
+ this.zkc = zkc;
+ basePath = conf.getZkLedgersRootPath() + "/underreplication";
+ electionPath = basePath + "/auditorelection";
+ createElectorPath();
+ }
+
+ /**
+ * Performing the auditor election using the ZooKeeper ephemeral sequential
+ * znode. The bookie which has created the least sequential will be elect as
+ * Auditor.
+ *
+ * @throws UnavailableException
+ * when performing auditor election
+ *
+ */
+ public void doElection() throws UnavailableException {
+ try {
+ // creating my vote in zk. Vote format is 'V_numeric'
+ createMyVote();
+ List<String> children = zkc.getChildren(getVotePath(""), false);
+
+ if (0 >= children.size()) {
+ throw new IllegalArgumentException(
+ "Atleast one bookie server should present to elect the Auditor!");
+ }
+
+ // sorting in ascending order of sequential number
+ Collections.sort(children, new ElectionComparator());
+ String voteNode = StringUtils.substringAfterLast(myVote,
+ PATH_SEPARATOR);
+
+ // starting Auditing service
+ if (children.get(AUDITOR_INDEX).equals(voteNode)) {
+ // update the auditor bookie id in the election path. This is
+ // done for debugging purpose
+ zkc.setData(getVotePath(""), bookieId.getBytes(), -1);
+ auditor = new Auditor(bookieId, conf, zkc);
+ auditor.start();
+ } else {
+ // If not an auditor, will be watching to my predecessor and
+ // looking the previous node deletion.
+ Watcher electionWatcher = new ElectionWatcher();
+ int myIndex = children.indexOf(voteNode);
+ int prevNodeIndex = myIndex - 1;
+ if (null == zkc.exists(getVotePath(PATH_SEPARATOR)
+ + children.get(prevNodeIndex), electionWatcher)) {
+ // While adding, the previous znode doesn't exists.
+ // Again going to election.
+ doElection();
+ }
+ }
+ } catch (KeeperException e) {
+ throw new UnavailableException(
+ "Exception while performing auditor election", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new UnavailableException(
+ "Interrupted while performing auditor election", e);
+ }
+ }
+
+ private void createMyVote() throws KeeperException, InterruptedException {
+ if (null == myVote || null == zkc.exists(myVote, false)) {
+ myVote = zkc.create(getVotePath(PATH_SEPARATOR + VOTE_PREFIX),
+ bookieId.getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.EPHEMERAL_SEQUENTIAL);
+ }
+ }
+
+ private String getVotePath(String vote) {
+ return electionPath + vote;
+ }
+
+ private void createElectorPath() throws UnavailableException {
+ try {
+ if (zkc.exists(basePath, false) == null) {
+ try {
+ zkc.create(basePath, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ } catch (KeeperException.NodeExistsException nee) {
+ // do nothing, someone else could have created it
+ }
+ }
+ if (zkc.exists(getVotePath(""), false) == null) {
+ try {
+ zkc.create(getVotePath(""), new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ } catch (KeeperException.NodeExistsException nee) {
+ // do nothing, someone else could have created it
+ }
+ }
+ } catch (KeeperException ke) {
+ throw new UnavailableException(
+ "Failed to initialize Auditor Elector", ke);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new UnavailableException(
+ "Failed to initialize Auditor Elector", ie);
+ }
+ }
+
+ /**
+ * Watching the predecessor bookies and will do election on predecessor node
+ * deletion or expiration.
+ */
+ private class ElectionWatcher implements Watcher {
+
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getType() == EventType.NodeDeleted) {
+ try {
+ doElection();
+ } catch (UnavailableException e) {
+ LOG.error("Exception when performing Auditor re-election",
+ e);
+ shutdown();
+ }
+ }
+ }
+ }
+
+ /**
+ * Shutting down AuditorElector
+ */
+ public void shutdown() {
+ if (!running) {
+ return;
+ }
+ running = false;
+ LOG.info("Shutting down AuditorElector");
+ try {
+ zkc.delete(myVote, -1);
+ } catch (InterruptedException ie) {
+ LOG.warn("InterruptedException while deleting myVote: " + myVote,
+ ie);
+ } catch (KeeperException ke) {
+ LOG.warn("Exception while deleting myVote:" + myVote, ke);
+ }
+ if (auditor != null) {
+ auditor.shutdown();
+ auditor = null;
+ }
+ }
+
+ /**
+ * If current bookie is running as auditor, return the status of the
+ * auditor. Otherwise return the status of elector.
+ *
+ * @return
+ */
+ public boolean isRunning() {
+ if (auditor != null) {
+ return auditor.isRunning();
+ }
+ return running;
+ }
+
+ /**
+ * Compare the votes in the ascending order of the sequence number. Vote
+ * format is 'V_sequencenumber', comparator will do sorting based on the
+ * numeric sequence value.
+ */
+ private static class ElectionComparator
+ implements Comparator<String>, Serializable {
+ /**
+ * Return -1 if the first vote is less than second. Return 1 if the
+ * first vote is greater than second. Return 0 if the votes are equal.
+ */
+ public int compare(String vote1, String vote2) {
+ long voteSeqId1 = getVoteSequenceId(vote1);
+ long voteSeqId2 = getVoteSequenceId(vote2);
+ int result = voteSeqId1 < voteSeqId2 ? -1
+ : (voteSeqId1 > voteSeqId2 ? 1 : 0);
+ return result;
+ }
+
+ private long getVoteSequenceId(String vote) {
+ String voteId = StringUtils.substringAfter(vote, VOTE_PREFIX);
+ return Long.parseLong(voteId);
+ }
+ }
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorBookieTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorBookieTest.java?rev=1377716&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorBookieTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorBookieTest.java Mon Aug 27 15:43:59 2012
@@ -0,0 +1,277 @@
+/**
+ *
+ * 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.bookkeeper.replication;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+
+import junit.framework.Assert;
+
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.util.StringUtils;
+import org.apache.zookeeper.KeeperException;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This test verifies the auditor bookie scenarios which will be monitoring the
+ * bookie failures
+ */
+public class AuditorBookieTest extends BookKeeperClusterTestCase {
+ // Depending on the taste, select the amount of logging
+ // by decommenting one of the two lines below
+ // static Logger LOG = Logger.getRootLogger();
+ private final static Logger LOG = LoggerFactory
+ .getLogger(AuditorBookieTest.class);
+ private String electionPath;
+ private HashMap<String, AuditorElector> auditorElectors = new HashMap<String, AuditorElector>();
+
+ public AuditorBookieTest() {
+ super(6);
+ electionPath = baseConf.getZkLedgersRootPath()
+ + "/underreplication/auditorelection";
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ startAuditorElectors();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ stopAuditorElectors();
+ super.tearDown();
+ }
+
+ /**
+ * Test should ensure only one should act as Auditor. Starting/shutdown
+ * other than auditor bookie shouldn't initiate re-election and multiple
+ * auditors.
+ */
+ @Test
+ public void testEnsureOnlySingleAuditor() throws Exception {
+ BookieServer auditor = verifyAuditor();
+
+ // shutdown bookie which is not an auditor
+ int indexOf = bs.indexOf(auditor);
+ int bkIndexDownBookie;
+ if (indexOf < bs.size() - 1) {
+ bkIndexDownBookie = indexOf + 1;
+ } else {
+ bkIndexDownBookie = indexOf - 1;
+ }
+ shudownBookie(bs.get(bkIndexDownBookie));
+
+ startNewBookie();
+ startNewBookie();
+ // grace period for the auditor re-election if any
+ BookieServer newAuditor = waitForNewAuditor(auditor);
+ Assert.assertSame(
+ "Auditor re-election is not happened for auditor failure!",
+ auditor, newAuditor);
+ }
+
+ /**
+ * Test Auditor crashes should trigger re-election and another bookie should
+ * take over the auditor ship
+ */
+ @Test
+ public void testSuccessiveAuditorCrashes() throws Exception {
+ BookieServer auditor = verifyAuditor();
+ shudownBookie(auditor);
+
+ BookieServer newAuditor1 = waitForNewAuditor(auditor);
+ bs.remove(auditor);
+
+ shudownBookie(newAuditor1);
+ BookieServer newAuditor2 = waitForNewAuditor(newAuditor1);
+ Assert.assertNotSame(
+ "Auditor re-election is not happened for auditor failure!",
+ auditor, newAuditor2);
+ bs.remove(newAuditor1);
+ }
+
+ /**
+ * Test restarting the entire bookie cluster. It shouldn't create multiple
+ * bookie auditors
+ */
+ @Test
+ public void testBookieClusterRestart() throws Exception {
+ BookieServer auditor = verifyAuditor();
+ for (AuditorElector auditorElector : auditorElectors.values()) {
+ assertTrue("Auditor elector is not running!", auditorElector
+ .isRunning());
+ }
+ stopBKCluster();
+ stopAuditorElectors();
+
+ startBKCluster();
+ startAuditorElectors();
+ BookieServer newAuditor = waitForNewAuditor(auditor);
+ Assert.assertNotSame(
+ "Auditor re-election is not happened for auditor failure!",
+ auditor, newAuditor);
+ }
+
+ /**
+ * Test the vote is deleting from the ZooKeeper during shutdown.
+ */
+ @Test
+ public void testShutdown() throws Exception {
+ BookieServer auditor = verifyAuditor();
+ shudownBookie(auditor);
+
+ // waiting for new auditor
+ BookieServer newAuditor = waitForNewAuditor(auditor);
+ Assert.assertNotSame(
+ "Auditor re-election is not happened for auditor failure!",
+ auditor, newAuditor);
+ int indexOfDownBookie = bs.indexOf(auditor);
+ bs.remove(indexOfDownBookie);
+ bsConfs.remove(indexOfDownBookie);
+ tmpDirs.remove(indexOfDownBookie);
+ List<String> children = zkc.getChildren(electionPath, false);
+ for (String child : children) {
+ byte[] data = zkc.getData(electionPath + '/' + child, false, null);
+ String bookieIP = new String(data);
+ StringBuilder addr = new StringBuilder();
+ StringUtils.addrToString(addr, auditor.getLocalAddress());
+ Assert.assertFalse("AuditorElection cleanup fails", bookieIP
+ .contains(addr));
+ }
+ }
+
+ /**
+ * Test restart of the previous Auditor bookie shouldn't initiate
+ * re-election and should create new vote after restarting.
+ */
+ @Test
+ public void testRestartAuditorBookieAfterCrashing() throws Exception {
+ BookieServer auditor = verifyAuditor();
+
+ shudownBookie(auditor);
+
+ // restarting Bookie with same configurations.
+ int indexOfDownBookie = bs.indexOf(auditor);
+ ServerConfiguration serverConfiguration = bsConfs
+ .get(indexOfDownBookie);
+ bs.remove(indexOfDownBookie);
+ bsConfs.remove(indexOfDownBookie);
+ tmpDirs.remove(indexOfDownBookie);
+ startBookie(serverConfiguration);
+ // starting corresponding auditor elector
+ StringBuilder sb = new StringBuilder();
+ StringUtils.addrToString(sb, auditor.getLocalAddress());
+ LOG.debug("Performing Auditor Election:" + sb.toString());
+ auditorElectors.get(sb.toString()).doElection();
+
+ // waiting for new auditor to come
+ BookieServer newAuditor = waitForNewAuditor(auditor);
+ Assert.assertNotSame(
+ "Auditor re-election is not happened for auditor failure!",
+ auditor, newAuditor);
+ Assert.assertFalse("No relection after old auditor rejoins", auditor
+ .getLocalAddress().getPort() == newAuditor.getLocalAddress()
+ .getPort());
+ }
+
+ private void startAuditorElectors() throws UnavailableException {
+ for (BookieServer bserver : bs) {
+ StringBuilder sb = new StringBuilder();
+ StringBuilder addr = StringUtils.addrToString(sb, bserver
+ .getLocalAddress());
+ AuditorElector auditorElector = new AuditorElector(addr.toString(),
+ baseClientConf, zkc);
+ auditorElectors.put(addr.toString(), auditorElector);
+ auditorElector.doElection();
+ LOG.debug("Starting Auditor Elector");
+ }
+ }
+
+ private void stopAuditorElectors() {
+ for (AuditorElector auditorElector : auditorElectors.values()) {
+ auditorElector.shutdown();
+ LOG.debug("Stopping Auditor Elector!");
+ }
+ }
+
+ private BookieServer verifyAuditor() throws KeeperException,
+ InterruptedException {
+ List<BookieServer> auditors = getAuditorBookie();
+ Assert.assertEquals("Multiple Bookies acting as Auditor!", 1, auditors
+ .size());
+ LOG.debug("Bookie running as Auditor:" + auditors.get(0));
+ return auditors.get(0);
+ }
+
+ private List<BookieServer> getAuditorBookie() throws KeeperException,
+ InterruptedException {
+ List<BookieServer> auditors = new LinkedList<BookieServer>();
+ byte[] data = zkc.getData(electionPath, false, null);
+ Assert.assertNotNull("Auditor election failed", data);
+ for (BookieServer bks : bs) {
+ if (new String(data).contains(bks.getLocalAddress().getPort() + "")) {
+ auditors.add(bks);
+ }
+ }
+ return auditors;
+ }
+
+ private void shudownBookie(BookieServer bkServer) {
+ StringBuilder sb = new StringBuilder();
+ StringUtils.addrToString(sb, bkServer.getLocalAddress());
+ LOG.debug("Shutting down bookie:" + sb.toString());
+
+ // shutdown bookie which is an auditor
+ bkServer.shutdown();
+ // stopping corresponding auditor elector
+ auditorElectors.get(sb.toString()).shutdown();
+ }
+
+ private BookieServer waitForNewAuditor(BookieServer auditor)
+ throws InterruptedException, KeeperException {
+ BookieServer newAuditor = null;
+ int retryCount = 8;
+ while (retryCount > 0) {
+ List<BookieServer> auditors = getAuditorBookie();
+ if (auditors.size() > 0) {
+ newAuditor = auditors.get(0);
+ if (auditor == newAuditor) {
+ Thread.sleep(500);
+ } else {
+ break;
+ }
+ }
+ retryCount--;
+ }
+ Assert.assertNotNull(
+ "New Auditor is not reelected after auditor crashes",
+ newAuditor);
+ verifyAuditor();
+ return newAuditor;
+ }
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java?rev=1377716&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java Mon Aug 27 15:43:59 2012
@@ -0,0 +1,352 @@
+/**
+ *
+ * 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.bookkeeper.replication;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
+import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
+import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
+import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.bookkeeper.test.MultiLedgerManagerTestCase;
+import org.apache.bookkeeper.util.StringUtils;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests publishing of under replicated ledgers by the Auditor bookie node when
+ * corresponding bookies identifes as not running
+ */
+public class AuditorLedgerCheckerTest extends MultiLedgerManagerTestCase {
+
+ // Depending on the taste, select the amount of logging
+ // by decommenting one of the two lines below
+ // static Logger LOG = Logger.getRootLogger();
+ private final static Logger LOG = LoggerFactory
+ .getLogger(AuditorLedgerCheckerTest.class);
+
+ private static final byte[] ledgerPassword = "aaa".getBytes();
+ private Random rng; // Random Number Generator
+
+ private DigestType digestType;
+
+ private final String UNDERREPLICATED_PATH = baseClientConf
+ .getZkLedgersRootPath()
+ + "/underreplication/ledgers";
+ private HashMap<String, AuditorElector> auditorElectors = new HashMap<String, AuditorElector>();
+ private LedgerUnderreplicationManager urLedgerMgr;
+ private Set<Long> urLedgerList;
+ private Map<Long, String> urLedgerData;
+ private List<Long> ledgerList;
+
+ public AuditorLedgerCheckerTest(String ledgerManagerFactoryClass)
+ throws IOException, KeeperException, InterruptedException,
+ CompatibilityException {
+ super(3);
+ LOG.info("Running test case using ledger manager : "
+ + ledgerManagerFactoryClass);
+ this.digestType = DigestType.CRC32;
+ // set ledger manager name
+ baseConf.setLedgerManagerFactoryClassName(ledgerManagerFactoryClass);
+ baseClientConf
+ .setLedgerManagerFactoryClassName(ledgerManagerFactoryClass);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ urLedgerMgr = new ZkLedgerUnderreplicationManager(baseClientConf, zkc);
+ startAuditorElectors();
+ rng = new Random(System.currentTimeMillis()); // Initialize the Random
+ urLedgerList = new HashSet<Long>();
+ urLedgerData = new HashMap<Long, String>();
+ ledgerList = new ArrayList<Long>(2);
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ stopAuditorElectors();
+ }
+
+ private void startAuditorElectors() throws UnavailableException {
+ for (BookieServer bserver : bs) {
+ StringBuilder sb = new StringBuilder();
+ StringBuilder addr = StringUtils.addrToString(sb, bserver
+ .getLocalAddress());
+ AuditorElector auditorElector = new AuditorElector(addr.toString(),
+ baseClientConf, zkc);
+ auditorElectors.put(addr.toString(), auditorElector);
+ auditorElector.doElection();
+ LOG.debug("Starting Auditor Elector");
+ }
+ }
+
+ private void stopAuditorElectors() {
+ for (AuditorElector auditorElector : auditorElectors.values()) {
+ auditorElector.shutdown();
+ LOG.debug("Stopping Auditor Elector!");
+ }
+ }
+
+ /**
+ * Test publishing of under replicated ledgers by the auditor bookie
+ */
+ @Test
+ public void testSimpleLedger() throws Exception {
+ LedgerHandle lh1 = createAndAddEntriesToLedger();
+ Long ledgerId = lh1.getId();
+ LOG.debug("Created ledger : " + ledgerId);
+ ledgerList.add(ledgerId);
+ lh1.close();
+
+ final CountDownLatch underReplicaLatch = registerUrLedgerWatcher(ledgerList
+ .size());
+
+ int bkShutdownIndex = bs.size() - 1;
+ String shutdownBookie = shutdownBookie(bkShutdownIndex);
+
+ // grace period for publishing the bk-ledger
+ LOG.debug("Waiting for ledgers to be marked as under replicated");
+ underReplicaLatch.await(5, TimeUnit.SECONDS);
+
+ assertEquals("Missed identifying under replicated ledgers", 1,
+ urLedgerList.size());
+
+ /*
+ * Sample data format present in the under replicated ledger path
+ *
+ * {4=replica: "10.18.89.153:5002"}
+ */
+ assertTrue("Ledger is not marked as underreplicated:" + ledgerId,
+ urLedgerList.contains(ledgerId));
+ String data = urLedgerData.get(ledgerId);
+ assertTrue("Bookie " + shutdownBookie
+ + "is not listed in the ledger as missing replica :" + data,
+ data.contains(shutdownBookie));
+ }
+
+ /**
+ * Test once published under replicated ledger should exists even after
+ * restarting respective bookie
+ */
+ @Test
+ public void testRestartBookie() throws Exception {
+ LedgerHandle lh1 = createAndAddEntriesToLedger();
+ ledgerList.add(lh1.getId());
+ LedgerHandle lh2 = createAndAddEntriesToLedger();
+ ledgerList.add(lh2.getId());
+ LOG.debug("Created following ledgers : " + ledgerList);
+
+ // 2 is added to the latch, since after the ledger reformation, again
+ // the reformed bookie is stopped. So auditor will modify the zk
+ // underreplicated metadata
+ int count = ledgerList.size() + 2;
+ final CountDownLatch underReplicaLatch = registerUrLedgerWatcher(count);
+
+ int bkShutdownIndex = bs.size() - 1;
+ ServerConfiguration bookieConf1 = bsConfs.get(bkShutdownIndex);
+ String shutdownBookie = shutdownBookie(bkShutdownIndex);
+
+ // restart the failed bookie and simulate previously listed ledgers are
+ // rereplicated
+ bs.add(startBookie(bookieConf1));
+
+ // grace period for publishing the bk-ledger
+ LOG.debug("Waiting for ledgers to be marked as under replicated");
+ underReplicaLatch.await(5, TimeUnit.SECONDS);
+
+ assertEquals("Missed identifying under replicated ledgers", 2,
+ urLedgerList.size());
+
+ /*
+ * Sample data format present in the under replicated ledger path
+ *
+ * {4=replica: "10.18.89.153:5002", 5=replica: "10.18.89.153:5003"}
+ */
+ for (Long ledgerId : ledgerList) {
+ assertTrue("Ledger is not marked as underreplicated:" + ledgerId,
+ urLedgerList.contains(ledgerId));
+ String data = urLedgerData.get(ledgerId);
+ assertTrue("Bookie " + shutdownBookie
+ + " is not listed in the ledger as missing " + data, data
+ .contains(shutdownBookie));
+ }
+ }
+
+ /**
+ * Test publishing of under replicated ledgers when multiple bookie failures
+ * one after another.
+ */
+ @Test
+ public void testMultipleBookieFailures() throws Exception {
+ LedgerHandle lh1 = createAndAddEntriesToLedger();
+ ledgerList.add(lh1.getId());
+ LedgerHandle lh2 = createAndAddEntriesToLedger();
+ ledgerList.add(lh2.getId());
+ LOG.debug("Created following ledgers : " + ledgerList);
+
+ // failing first bookie
+ shutdownBookie(bs.size() - 1);
+ // simulate re-replication
+ doLedgerRereplication(lh1.getId());
+ doLedgerRereplication(lh2.getId());
+
+ // failing another bookie
+ CountDownLatch underReplicaLatch = registerUrLedgerWatcher(ledgerList
+ .size());
+ String shutdownBookie = shutdownBookie(bs.size() - 1);
+
+ // grace period for publishing the bk-ledger
+ LOG.debug("Waiting for ledgers to be marked as under replicated");
+ underReplicaLatch.await(5, TimeUnit.SECONDS);
+
+ assertEquals("Missed identifying under replicated ledgers", 2,
+ urLedgerList.size());
+
+ /*
+ * Sample data format present in the under replicated ledger path
+ * {4=replica: "10.18.89.153:5002", 5=replica: "10.18.89.153:5003"}
+ */
+ for (Long ledgerId : ledgerList) {
+ assertTrue("Ledger is not marked as underreplicated:" + ledgerId,
+ urLedgerList.contains(ledgerId));
+ String data = urLedgerData.get(ledgerId);
+ assertTrue("Bookie " + shutdownBookie
+ + " is not listed in the ledger as missing " + data, data
+ .contains(shutdownBookie));
+ }
+ }
+
+ private CountDownLatch registerUrLedgerWatcher(int count)
+ throws KeeperException, InterruptedException {
+ final CountDownLatch underReplicaLatch = new CountDownLatch(count);
+ for (Long ledgerId : ledgerList) {
+ Watcher urLedgerWatcher = new ChildWatcher(underReplicaLatch);
+ String znode = String.format("%s/urL%010d", getParentZnodePath(
+ UNDERREPLICATED_PATH, ledgerId), ledgerId);
+ zkc.exists(znode, urLedgerWatcher);
+ }
+ return underReplicaLatch;
+ }
+
+ private void doLedgerRereplication(long ledgerId)
+ throws UnavailableException {
+ urLedgerMgr.getLedgerToRereplicate();
+ urLedgerMgr.markLedgerReplicated(ledgerId);
+ urLedgerMgr.releaseUnderreplicatedLedger(ledgerId);
+ urLedgerData.clear();
+ }
+
+ private String shutdownBookie(int bkShutdownIndex) throws IOException,
+ InterruptedException {
+ StringBuilder bookieAddr = new StringBuilder();
+ BookieServer bkServer = bs.get(bkShutdownIndex);
+ StringUtils.addrToString(bookieAddr, bkServer.getLocalAddress());
+ LOG.debug("Shutting down bookie:" + bookieAddr.toString());
+ killBookie(bkShutdownIndex);
+ auditorElectors.get(bookieAddr.toString()).shutdown();
+ auditorElectors.remove(bookieAddr.toString());
+ return bookieAddr.toString();
+ }
+
+ private LedgerHandle createAndAddEntriesToLedger() throws BKException,
+ InterruptedException {
+ int numEntriesToWrite = 100;
+ // Create a ledger
+ LedgerHandle lh = bkc.createLedger(digestType, ledgerPassword);
+ LOG.info("Ledger ID: " + lh.getId());
+ addEntry(numEntriesToWrite, lh);
+ return lh;
+ }
+
+ private void addEntry(int numEntriesToWrite, LedgerHandle lh)
+ throws InterruptedException, BKException {
+ for (int i = 0; i < numEntriesToWrite; i++) {
+ ByteBuffer entry = ByteBuffer.allocate(4);
+ entry.putInt(rng.nextInt(Integer.MAX_VALUE));
+ entry.position(0);
+ lh.addEntry(entry.array());
+ }
+ }
+
+ private String getParentZnodePath(String base, long ledgerId) {
+ String subdir1 = String.format("%04x", ledgerId >> 48 & 0xffff);
+ String subdir2 = String.format("%04x", ledgerId >> 32 & 0xffff);
+ String subdir3 = String.format("%04x", ledgerId >> 16 & 0xffff);
+ String subdir4 = String.format("%04x", ledgerId & 0xffff);
+
+ return String.format("%s/%s/%s/%s/%s", base, subdir1, subdir2, subdir3,
+ subdir4);
+ }
+
+ private class ChildWatcher implements Watcher {
+ private final CountDownLatch underReplicaLatch;
+
+ public ChildWatcher(CountDownLatch underReplicaLatch) {
+ this.underReplicaLatch = underReplicaLatch;
+ }
+
+ @Override
+ public void process(WatchedEvent event) {
+ LOG.debug("Recieved notification for the ledger path : "
+ + event.getPath());
+ for (Long ledgerId : ledgerList) {
+ if (event.getPath().contains(ledgerId + "")) {
+ urLedgerList.add(Long.valueOf(ledgerId));
+ try {
+ byte[] data = zkc.getData(event.getPath(), this, null);
+ urLedgerData.put(ledgerId, new String(data));
+ } catch (KeeperException e) {
+ LOG.error("Exception while reading data from znode :"
+ + event.getPath());
+ } catch (InterruptedException e) {
+ LOG.error("Exception while reading data from znode :"
+ + event.getPath());
+ }
+ }
+ }
+ LOG.debug("Count down and waiting for next notification");
+ // count down and waiting for next notification
+ underReplicaLatch.countDown();
+ }
+ }
+}