You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by en...@apache.org on 2013/02/22 01:15:53 UTC
svn commit: r1448867 [2/3] - in /hbase/trunk:
hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/
hbase-protocol/src/main/protobuf/
hbase-server/src/main/java/org/apache/hadoop/hbase/
hbase-server/src/main/java/org/apache/hadoop/hb...
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java Fri Feb 22 00:15:52 2013
@@ -41,6 +41,8 @@ import org.apache.hadoop.hbase.master.Ma
import org.apache.hadoop.hbase.master.RegionPlan;
import org.apache.hadoop.hbase.master.RegionStates;
import org.apache.hadoop.hbase.master.ServerManager;
+import org.apache.hadoop.hbase.master.TableLockManager;
+import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.zookeeper.KeeperException;
@@ -55,41 +57,60 @@ public class EnableTableHandler extends
private final byte [] tableName;
private final String tableNameStr;
private final AssignmentManager assignmentManager;
- private final CatalogTracker ct;
+ private final TableLockManager tableLockManager;
+ private final CatalogTracker catalogTracker;
private boolean retainAssignment = false;
+ private TableLock tableLock;
public EnableTableHandler(Server server, byte [] tableName,
CatalogTracker catalogTracker, AssignmentManager assignmentManager,
- boolean skipTableStateCheck)
- throws TableNotFoundException, TableNotDisabledException, IOException {
+ TableLockManager tableLockManager, boolean skipTableStateCheck) {
super(server, EventType.C_M_ENABLE_TABLE);
this.tableName = tableName;
this.tableNameStr = Bytes.toString(tableName);
- this.ct = catalogTracker;
+ this.catalogTracker = catalogTracker;
this.assignmentManager = assignmentManager;
+ this.tableLockManager = tableLockManager;
this.retainAssignment = skipTableStateCheck;
- // Check if table exists
- if (!MetaReader.tableExists(catalogTracker, this.tableNameStr)) {
- throw new TableNotFoundException(Bytes.toString(tableName));
- }
+ }
- // There could be multiple client requests trying to disable or enable
- // the table at the same time. Ensure only the first request is honored
- // After that, no other requests can be accepted until the table reaches
- // DISABLED or ENABLED.
- if (!skipTableStateCheck)
- {
- try {
- if (!this.assignmentManager.getZKTable().checkDisabledAndSetEnablingTable
- (this.tableNameStr)) {
- LOG.info("Table " + tableNameStr + " isn't disabled; skipping enable");
- throw new TableNotDisabledException(this.tableNameStr);
- }
- } catch (KeeperException e) {
- throw new IOException("Unable to ensure that the table will be" +
- " enabling because of a ZooKeeper issue", e);
+ public EnableTableHandler prepare()
+ throws TableNotFoundException, TableNotDisabledException, IOException {
+ //acquire the table write lock, blocking
+ this.tableLock = this.tableLockManager.writeLock(tableName,
+ EventType.C_M_ENABLE_TABLE.toString());
+ this.tableLock.acquire();
+
+ boolean success = false;
+ try {
+ // Check if table exists
+ if (!MetaReader.tableExists(catalogTracker, this.tableNameStr)) {
+ throw new TableNotFoundException(Bytes.toString(tableName));
+ }
+
+ // There could be multiple client requests trying to disable or enable
+ // the table at the same time. Ensure only the first request is honored
+ // After that, no other requests can be accepted until the table reaches
+ // DISABLED or ENABLED.
+ if (!retainAssignment) {
+ try {
+ if (!this.assignmentManager.getZKTable().checkDisabledAndSetEnablingTable
+ (this.tableNameStr)) {
+ LOG.info("Table " + tableNameStr + " isn't disabled; skipping enable");
+ throw new TableNotDisabledException(this.tableNameStr);
+ }
+ } catch (KeeperException e) {
+ throw new IOException("Unable to ensure that the table will be" +
+ " enabling because of a ZooKeeper issue", e);
+ }
+ }
+ success = true;
+ } finally {
+ if (!success) {
+ releaseTableLock();
}
}
+ return this;
}
@Override
@@ -121,6 +142,18 @@ public class EnableTableHandler extends
LOG.error("Error trying to enable the table " + this.tableNameStr, e);
} catch (InterruptedException e) {
LOG.error("Error trying to enable the table " + this.tableNameStr, e);
+ } finally {
+ releaseTableLock();
+ }
+ }
+
+ private void releaseTableLock() {
+ if (this.tableLock != null) {
+ try {
+ this.tableLock.release();
+ } catch (IOException ex) {
+ LOG.warn("Could not release the table lock", ex);
+ }
}
}
@@ -134,7 +167,7 @@ public class EnableTableHandler extends
// Get the regions of this table. We're done when all listed
// tables are onlined.
List<Pair<HRegionInfo, ServerName>> tableRegionsAndLocations = MetaReader
- .getTableRegionsAndLocations(this.ct, tableName, true);
+ .getTableRegionsAndLocations(this.catalogTracker, tableName, true);
int countOfRegionsInTable = tableRegionsAndLocations.size();
List<HRegionInfo> regions = regionsToAssignWithServerName(tableRegionsAndLocations);
int regionsCount = regions.size();
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/ModifyTableHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/ModifyTableHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/ModifyTableHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/ModifyTableHandler.java Fri Feb 22 00:15:52 2013
@@ -35,16 +35,20 @@ public class ModifyTableHandler extends
public ModifyTableHandler(final byte [] tableName,
final HTableDescriptor htd, final Server server,
- final MasterServices masterServices)
- throws IOException {
+ final MasterServices masterServices) {
super(EventType.C_M_MODIFY_TABLE, tableName, server, masterServices);
- // Check table exists.
- getTableDescriptor();
// This is the new schema we are going to write out as this modification.
this.htd = htd;
}
@Override
+ protected void prepareWithTableLock() throws IOException {
+ super.prepareWithTableLock();
+ // Check table exists.
+ getTableDescriptor();
+ }
+
+ @Override
protected void handleTableOperation(List<HRegionInfo> hris)
throws IOException {
MasterCoprocessorHost cpHost = ((HMaster) this.server)
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableAddFamilyHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableAddFamilyHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableAddFamilyHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableAddFamilyHandler.java Fri Feb 22 00:15:52 2013
@@ -40,14 +40,19 @@ public class TableAddFamilyHandler exten
private final HColumnDescriptor familyDesc;
public TableAddFamilyHandler(byte[] tableName, HColumnDescriptor familyDesc,
- Server server, final MasterServices masterServices) throws IOException {
+ Server server, final MasterServices masterServices) {
super(EventType.C_M_ADD_FAMILY, tableName, server, masterServices);
+ this.familyDesc = familyDesc;
+ }
+
+ @Override
+ protected void prepareWithTableLock() throws IOException {
+ super.prepareWithTableLock();
HTableDescriptor htd = getTableDescriptor();
if (htd.hasFamily(familyDesc.getName())) {
throw new InvalidFamilyOperationException("Family '" +
familyDesc.getNameAsString() + "' already exists so cannot be added");
}
- this.familyDesc = familyDesc;
}
@Override
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableDeleteFamilyHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableDeleteFamilyHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableDeleteFamilyHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableDeleteFamilyHandler.java Fri Feb 22 00:15:52 2013
@@ -37,11 +37,17 @@ import org.apache.hadoop.hbase.util.Byte
@InterfaceAudience.Private
public class TableDeleteFamilyHandler extends TableEventHandler {
- private final byte [] familyName;
+ private byte [] familyName;
public TableDeleteFamilyHandler(byte[] tableName, byte [] familyName,
Server server, final MasterServices masterServices) throws IOException {
- super(EventType.C_M_ADD_FAMILY, tableName, server, masterServices);
+ super(EventType.C_M_DELETE_FAMILY, tableName, server, masterServices);
+ this.familyName = familyName;
+ }
+
+ @Override
+ protected void prepareWithTableLock() throws IOException {
+ super.prepareWithTableLock();
HTableDescriptor htd = getTableDescriptor();
this.familyName = hasColumnFamily(htd, familyName);
}
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableEventHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableEventHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableEventHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableEventHandler.java Fri Feb 22 00:15:52 2013
@@ -42,6 +42,7 @@ import org.apache.hadoop.hbase.client.HT
import org.apache.hadoop.hbase.executor.EventHandler;
import org.apache.hadoop.hbase.master.BulkReOpen;
import org.apache.hadoop.hbase.master.MasterServices;
+import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.zookeeper.KeeperException;
@@ -61,37 +62,61 @@ public abstract class TableEventHandler
protected final MasterServices masterServices;
protected final byte [] tableName;
protected final String tableNameStr;
+ protected TableLock tableLock;
public TableEventHandler(EventType eventType, byte [] tableName, Server server,
- MasterServices masterServices)
- throws IOException {
+ MasterServices masterServices) {
super(server, eventType);
this.masterServices = masterServices;
this.tableName = tableName;
+ this.tableNameStr = Bytes.toString(this.tableName);
+ }
+
+ public TableEventHandler prepare() throws IOException {
+ //acquire the table write lock, blocking
+ this.tableLock = masterServices.getTableLockManager()
+ .writeLock(tableName, eventType.toString());
+ this.tableLock.acquire();
+ boolean success = false;
try {
- this.masterServices.checkTableModifiable(tableName);
- } catch (TableNotDisabledException ex) {
- if (isOnlineSchemaChangeAllowed()
- && eventType.isOnlineSchemaChangeSupported()) {
- LOG.debug("Ignoring table not disabled exception " +
- "for supporting online schema changes.");
- } else {
- throw ex;
+ try {
+ this.masterServices.checkTableModifiable(tableName);
+ } catch (TableNotDisabledException ex) {
+ if (isOnlineSchemaChangeAllowed()
+ && eventType.isOnlineSchemaChangeSupported()) {
+ LOG.debug("Ignoring table not disabled exception " +
+ "for supporting online schema changes.");
+ } else {
+ throw ex;
+ }
+ }
+ prepareWithTableLock();
+ success = true;
+ } finally {
+ if (!success ) {
+ releaseTableLock();
}
}
- this.tableNameStr = Bytes.toString(this.tableName);
+ return this;
}
- private boolean isOnlineSchemaChangeAllowed() {
- return this.server.getConfiguration().getBoolean(
- "hbase.online.schema.update.enable", false);
- }
+ /** Called from prepare() while holding the table lock. Subclasses
+ * can do extra initialization, and not worry about the releasing
+ * the table lock. */
+ protected void prepareWithTableLock() throws IOException {
+ }
+
+ private boolean isOnlineSchemaChangeAllowed() {
+ return this.server.getConfiguration().getBoolean(
+ "hbase.online.schema.update.enable", false);
+ }
@Override
public void process() {
try {
LOG.info("Handling table operation " + eventType + " on table " +
Bytes.toString(tableName));
+
List<HRegionInfo> hris =
MetaReader.getTableRegions(this.server.getCatalogTracker(),
tableName);
@@ -110,7 +135,19 @@ public abstract class TableEventHandler
LOG.error("Error manipulating table " + Bytes.toString(tableName), e);
} catch (KeeperException e) {
LOG.error("Error manipulating table " + Bytes.toString(tableName), e);
- }
+ } finally {
+ releaseTableLock();
+ }
+ }
+
+ protected void releaseTableLock() {
+ if (this.tableLock != null) {
+ try {
+ this.tableLock.release();
+ } catch (IOException ex) {
+ LOG.warn("Could not release the table lock", ex);
+ }
+ }
}
public boolean reOpenAllRegions(List<HRegionInfo> regions) throws IOException {
@@ -137,7 +174,7 @@ public abstract class TableEventHandler
reRegions.add(hri);
serverToRegions.get(rsLocation).add(hri);
}
-
+
LOG.info("Reopening " + reRegions.size() + " regions on "
+ serverToRegions.size() + " region servers.");
this.masterServices.getAssignmentManager().setRegionsToReopen(reRegions);
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableModifyFamilyHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableModifyFamilyHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableModifyFamilyHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/handler/TableModifyFamilyHandler.java Fri Feb 22 00:15:52 2013
@@ -25,12 +25,10 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.InvalidFamilyOperationException;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
import org.apache.hadoop.hbase.master.MasterServices;
-import org.apache.hadoop.hbase.util.Bytes;
/**
* Handles adding a new family to an existing table.
@@ -41,11 +39,16 @@ public class TableModifyFamilyHandler ex
public TableModifyFamilyHandler(byte[] tableName,
HColumnDescriptor familyDesc, Server server,
- final MasterServices masterServices) throws IOException {
+ final MasterServices masterServices) {
super(EventType.C_M_MODIFY_FAMILY, tableName, server, masterServices);
+ this.familyDesc = familyDesc;
+ }
+
+ @Override
+ protected void prepareWithTableLock() throws IOException {
+ super.prepareWithTableLock();
HTableDescriptor htd = getTableDescriptor();
hasColumnFamily(htd, familyDesc.getName());
- this.familyDesc = familyDesc;
}
@Override
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/CloneSnapshotHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/CloneSnapshotHandler.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/CloneSnapshotHandler.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/CloneSnapshotHandler.java Fri Feb 22 00:15:52 2013
@@ -32,7 +32,6 @@ import org.apache.hadoop.hbase.HRegionIn
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NotAllMetaRegionsOnlineException;
import org.apache.hadoop.hbase.TableExistsException;
-import org.apache.hadoop.hbase.catalog.MetaReader;
import org.apache.hadoop.hbase.errorhandling.ForeignException;
import org.apache.hadoop.hbase.errorhandling.ForeignExceptionDispatcher;
import org.apache.hadoop.hbase.master.MasterServices;
@@ -42,7 +41,6 @@ import org.apache.hadoop.hbase.protobuf.
import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
-import org.apache.hadoop.hbase.util.Bytes;
import com.google.common.base.Preconditions;
@@ -68,8 +66,7 @@ public class CloneSnapshotHandler extend
final SnapshotDescription snapshot, final HTableDescriptor hTableDescriptor)
throws NotAllMetaRegionsOnlineException, TableExistsException, IOException {
super(masterServices, masterServices.getMasterFileSystem(), hTableDescriptor,
- masterServices.getConfiguration(), null, masterServices.getCatalogTracker(),
- masterServices.getAssignmentManager());
+ masterServices.getConfiguration(), null, masterServices);
// Snapshot information
this.snapshot = snapshot;
Added: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/DeletionListener.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/DeletionListener.java?rev=1448867&view=auto
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/DeletionListener.java (added)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/DeletionListener.java Fri Feb 22 00:15:52 2013
@@ -0,0 +1,101 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.zookeeper;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * A ZooKeeper watcher meant to detect deletions of ZNodes.
+ */
+@InterfaceAudience.Private
+public class DeletionListener extends ZooKeeperListener {
+
+ private static final Log LOG = LogFactory.getLog(DeletionListener.class);
+
+ private final String pathToWatch;
+ private final CountDownLatch deletedLatch;
+
+ private volatile Throwable exception;
+
+ /**
+ * Create a new instance of the deletion watcher.
+ * @param zkWatcher ZookeeperWatcher instance
+ * @param pathToWatch (Fully qualified) ZNode path that we are waiting to
+ * be deleted.
+ * @param deletedLatch Count down on this latch when deletion has occured.
+ */
+ public DeletionListener(ZooKeeperWatcher zkWatcher, String pathToWatch,
+ CountDownLatch deletedLatch) {
+ super(zkWatcher);
+ this.pathToWatch = pathToWatch;
+ this.deletedLatch = deletedLatch;
+ exception = null;
+ }
+
+ /**
+ * Check if an exception has occurred when re-setting the watch.
+ * @return True if we were unable to re-set a watch on a ZNode due to
+ * an exception.
+ */
+ public boolean hasException() {
+ return exception != null;
+ }
+
+ /**
+ * Get the last exception which has occurred when re-setting the watch.
+ * Use hasException() to check whether or not an exception has occurred.
+ * @return The last exception observed when re-setting the watch.
+ */
+ public Throwable getException() {
+ return exception;
+ }
+
+ @Override
+ public void nodeDataChanged(String path) {
+ if (!path.equals(pathToWatch)) {
+ return;
+ }
+ try {
+ if (!(ZKUtil.setWatchIfNodeExists(watcher, pathToWatch))) {
+ deletedLatch.countDown();
+ }
+ } catch (KeeperException ex) {
+ exception = ex;
+ deletedLatch.countDown();
+ LOG.error("Error when re-setting the watch on " + pathToWatch, ex);
+ }
+ }
+
+ @Override
+ public void nodeDeleted(String path) {
+ if (!path.equals(pathToWatch)) {
+ return;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Processing delete on " + pathToWatch);
+ }
+ deletedLatch.countDown();
+ }
+}
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/RecoverableZooKeeper.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/RecoverableZooKeeper.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/RecoverableZooKeeper.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/RecoverableZooKeeper.java Fri Feb 22 00:15:52 2013
@@ -49,24 +49,24 @@ import org.apache.zookeeper.proto.SetDat
/**
* A zookeeper that can handle 'recoverable' errors.
- * To handle recoverable errors, developers need to realize that there are two
- * classes of requests: idempotent and non-idempotent requests. Read requests
- * and unconditional sets and deletes are examples of idempotent requests, they
- * can be reissued with the same results.
- * (Although, the delete may throw a NoNodeException on reissue its effect on
- * the ZooKeeper state is the same.) Non-idempotent requests need special
- * handling, application and library writers need to keep in mind that they may
- * need to encode information in the data or name of znodes to detect
- * retries. A simple example is a create that uses a sequence flag.
- * If a process issues a create("/x-", ..., SEQUENCE) and gets a connection
- * loss exception, that process will reissue another
- * create("/x-", ..., SEQUENCE) and get back x-111. When the process does a
- * getChildren("/"), it sees x-1,x-30,x-109,x-110,x-111, now it could be
- * that x-109 was the result of the previous create, so the process actually
- * owns both x-109 and x-111. An easy way around this is to use "x-process id-"
+ * To handle recoverable errors, developers need to realize that there are two
+ * classes of requests: idempotent and non-idempotent requests. Read requests
+ * and unconditional sets and deletes are examples of idempotent requests, they
+ * can be reissued with the same results.
+ * (Although, the delete may throw a NoNodeException on reissue its effect on
+ * the ZooKeeper state is the same.) Non-idempotent requests need special
+ * handling, application and library writers need to keep in mind that they may
+ * need to encode information in the data or name of znodes to detect
+ * retries. A simple example is a create that uses a sequence flag.
+ * If a process issues a create("/x-", ..., SEQUENCE) and gets a connection
+ * loss exception, that process will reissue another
+ * create("/x-", ..., SEQUENCE) and get back x-111. When the process does a
+ * getChildren("/"), it sees x-1,x-30,x-109,x-110,x-111, now it could be
+ * that x-109 was the result of the previous create, so the process actually
+ * owns both x-109 and x-111. An easy way around this is to use "x-process id-"
* when doing the create. If the process is using an id of 352, before reissuing
- * the create it will do a getChildren("/") and see "x-222-1", "x-542-30",
- * "x-352-109", x-333-110". The process will know that the original create
+ * the create it will do a getChildren("/") and see "x-222-1", "x-542-30",
+ * "x-352-109", x-333-110". The process will know that the original create
* succeeded an the znode it created is "x-352-109".
* @see "http://wiki.apache.org/hadoop/ZooKeeper/ErrorHandling"
*/
@@ -99,15 +99,25 @@ public class RecoverableZooKeeper {
private static final int ID_LENGTH_SIZE = Bytes.SIZEOF_INT;
public RecoverableZooKeeper(String quorumServers, int sessionTimeout,
- Watcher watcher, int maxRetries, int retryIntervalMillis)
+ Watcher watcher, int maxRetries, int retryIntervalMillis)
+ throws IOException {
+ this(quorumServers, sessionTimeout, watcher, maxRetries, retryIntervalMillis,
+ null);
+ }
+
+ public RecoverableZooKeeper(String quorumServers, int sessionTimeout,
+ Watcher watcher, int maxRetries, int retryIntervalMillis, String identifier)
throws IOException {
this.zk = new ZooKeeper(quorumServers, sessionTimeout, watcher);
this.retryCounterFactory =
new RetryCounterFactory(maxRetries, retryIntervalMillis);
- // the identifier = processID@hostName
- this.identifier = ManagementFactory.getRuntimeMXBean().getName();
+ if (identifier == null || identifier.length() == 0) {
+ // the identifier = processID@hostName
+ identifier = ManagementFactory.getRuntimeMXBean().getName();
+ }
LOG.info("The identifier of this process is " + identifier);
+ this.identifier = identifier;
this.id = Bytes.toBytes(identifier);
this.watcher = watcher;
@@ -343,7 +353,7 @@ public class RecoverableZooKeeper {
/**
* setData is NOT an idempotent operation. Retry may cause BadVersion Exception
- * Adding an identifier field into the data to check whether
+ * Adding an identifier field into the data to check whether
* badversion is caused by the result of previous correctly setData
* @return Stat instance
*/
@@ -390,17 +400,17 @@ public class RecoverableZooKeeper {
/**
* <p>
- * NONSEQUENTIAL create is idempotent operation.
+ * NONSEQUENTIAL create is idempotent operation.
* Retry before throwing exceptions.
* But this function will not throw the NodeExist exception back to the
* application.
* </p>
* <p>
- * But SEQUENTIAL is NOT idempotent operation. It is necessary to add
- * identifier to the path to verify, whether the previous one is successful
+ * But SEQUENTIAL is NOT idempotent operation. It is necessary to add
+ * identifier to the path to verify, whether the previous one is successful
* or not.
* </p>
- *
+ *
* @return Path
*/
public String create(String path, byte[] data, List<ACL> acl,
@@ -417,12 +427,12 @@ public class RecoverableZooKeeper {
return createSequential(path, newData, acl, createMode);
default:
- throw new IllegalArgumentException("Unrecognized CreateMode: " +
+ throw new IllegalArgumentException("Unrecognized CreateMode: " +
createMode);
}
}
- private String createNonSequential(String path, byte[] data, List<ACL> acl,
+ private String createNonSequential(String path, byte[] data, List<ACL> acl,
CreateMode createMode) throws KeeperException, InterruptedException {
RetryCounter retryCounter = retryCounterFactory.create();
boolean isRetry = false; // False for first attempt, true for all retries.
@@ -435,14 +445,14 @@ public class RecoverableZooKeeper {
if (isRetry) {
// If the connection was lost, there is still a possibility that
// we have successfully created the node at our previous attempt,
- // so we read the node and compare.
+ // so we read the node and compare.
byte[] currentData = zk.getData(path, false, null);
if (currentData != null &&
- Bytes.compareTo(currentData, data) == 0) {
+ Bytes.compareTo(currentData, data) == 0) {
// We successfully created a non-sequential node
return path;
}
- LOG.error("Node " + path + " already exists with " +
+ LOG.error("Node " + path + " already exists with " +
Bytes.toStringBinary(currentData) + ", could not write " +
Bytes.toStringBinary(data));
throw e;
@@ -466,8 +476,8 @@ public class RecoverableZooKeeper {
isRetry = true;
}
}
-
- private String createSequential(String path, byte[] data,
+
+ private String createSequential(String path, byte[] data,
List<ACL> acl, CreateMode createMode)
throws KeeperException, InterruptedException {
RetryCounter retryCounter = retryCounterFactory.create();
@@ -573,7 +583,7 @@ public class RecoverableZooKeeper {
}
return null;
}
-
+
public byte[] removeMetaData(byte[] data) {
if(data == null || data.length == 0) {
return data;
@@ -642,7 +652,7 @@ public class RecoverableZooKeeper {
* @param prefixes the prefixes to include in the result
* @return list of every element that starts with one of the prefixes
*/
- private static List<String> filterByPrefix(List<String> nodes,
+ private static List<String> filterByPrefix(List<String> nodes,
String... prefixes) {
List<String> lockChildren = new ArrayList<String>();
for (String child : nodes){
@@ -655,4 +665,8 @@ public class RecoverableZooKeeper {
}
return lockChildren;
}
+
+ public String getIdentifier() {
+ return identifier;
+ }
}
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java Fri Feb 22 00:15:52 2013
@@ -82,7 +82,7 @@ public class ZKUtil {
private static final Log LOG = LogFactory.getLog(ZKUtil.class);
// TODO: Replace this with ZooKeeper constant when ZOOKEEPER-277 is resolved.
- private static final char ZNODE_PATH_SEPARATOR = '/';
+ public static final char ZNODE_PATH_SEPARATOR = '/';
private static int zkDumpConnectionTimeOut;
/**
@@ -106,18 +106,18 @@ public class ZKUtil {
public static RecoverableZooKeeper connect(Configuration conf, String ensemble,
Watcher watcher)
throws IOException {
- return connect(conf, ensemble, watcher, "");
+ return connect(conf, ensemble, watcher, null);
}
public static RecoverableZooKeeper connect(Configuration conf, String ensemble,
- Watcher watcher, final String descriptor)
+ Watcher watcher, final String identifier)
throws IOException {
if(ensemble == null) {
throw new IOException("Unable to determine ZooKeeper ensemble");
}
int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT,
HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
- LOG.debug(descriptor + " opening connection to ZooKeeper with ensemble (" +
+ LOG.debug(identifier + " opening connection to ZooKeeper with ensemble (" +
ensemble + ")");
int retry = conf.getInt("zookeeper.recovery.retry", 3);
int retryIntervalMillis =
@@ -125,7 +125,7 @@ public class ZKUtil {
zkDumpConnectionTimeOut = conf.getInt("zookeeper.dump.connection.timeout",
1000);
return new RecoverableZooKeeper(ensemble, timeout, watcher,
- retry, retryIntervalMillis);
+ retry, retryIntervalMillis, identifier);
}
/**
@@ -437,6 +437,30 @@ public class ZKUtil {
}
/**
+ * Watch the specified znode, but only if exists. Useful when watching
+ * for deletions. Uses .getData() (and handles NoNodeException) instead
+ * of .exists() to accomplish this, as .getData() will only set a watch if
+ * the znode exists.
+ * @param zkw zk reference
+ * @param znode path of node to watch
+ * @return true if the watch is set, false if node does not exists
+ * @throws KeeperException if unexpected zookeeper exception
+ */
+ public static boolean setWatchIfNodeExists(ZooKeeperWatcher zkw, String znode)
+ throws KeeperException {
+ try {
+ zkw.getRecoverableZooKeeper().getData(znode, true, null);
+ return true;
+ } catch (NoNodeException e) {
+ return false;
+ } catch (InterruptedException e) {
+ LOG.warn(zkw.prefix("Unable to set watcher on znode " + znode), e);
+ zkw.interruptedException(e);
+ return false;
+ }
+ }
+
+ /**
* Check if the specified node exists. Sets no watches.
*
* @param zkw zk reference
@@ -525,15 +549,13 @@ public class ZKUtil {
/**
* Lists the children of the specified znode without setting any watches.
*
- * Used to list the currently online regionservers and their addresses.
- *
* Sets no watches at all, this method is best effort.
*
* Returns an empty list if the node has no children. Returns null if the
* parent node itself does not exist.
*
* @param zkw zookeeper reference
- * @param znode node to get children of as addresses
+ * @param znode node to get children
* @return list of data of children of specified znode, empty if no children,
* null if parent does not exist
* @throws KeeperException if unexpected zookeeper exception
@@ -1026,6 +1048,36 @@ public class ZKUtil {
}
/**
+ * Creates the specified znode with the specified data but does not watch it.
+ *
+ * Returns the znode of the newly created node
+ *
+ * If there is another problem, a KeeperException will be thrown.
+ *
+ * @param zkw zk reference
+ * @param znode path of node
+ * @param data data of node
+ * @param createMode specifying whether the node to be created is ephemeral and/or sequential
+ * @return true name of the newly created znode or null
+ * @throws KeeperException if unexpected zookeeper exception
+ */
+ public static String createNodeIfNotExistsNoWatch(ZooKeeperWatcher zkw, String znode,
+ byte[] data, CreateMode createMode) throws KeeperException {
+
+ String createdZNode = null;
+ try {
+ createdZNode = zkw.getRecoverableZooKeeper().create(znode, data,
+ createACL(zkw, znode), createMode);
+ } catch (KeeperException.NodeExistsException nee) {
+ return znode;
+ } catch (InterruptedException e) {
+ zkw.interruptedException(e);
+ return null;
+ }
+ return createdZNode;
+ }
+
+ /**
* Creates the specified node with the specified data and watches it.
*
* <p>Throws an exception if the node already exists.
@@ -1346,7 +1398,7 @@ public class ZKUtil {
CreateAndFailSilent op = (CreateAndFailSilent) o;
return getPath().equals(op.getPath()) && Arrays.equals(data, op.data);
}
-
+
@Override
public int hashCode() {
int ret = 17 + getPath().hashCode() * 31;
@@ -1370,7 +1422,7 @@ public class ZKUtil {
return super.equals(o);
}
-
+
@Override
public int hashCode() {
return getPath().hashCode();
@@ -1400,7 +1452,7 @@ public class ZKUtil {
SetData op = (SetData) o;
return getPath().equals(op.getPath()) && Arrays.equals(data, op.data);
}
-
+
@Override
public int hashCode() {
int ret = getPath().hashCode();
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java Fri Feb 22 00:15:52 2013
@@ -103,6 +103,8 @@ public class ZooKeeperWatcher implements
public String splitLogZNode;
// znode containing the state of the load balancer
public String balancerZNode;
+ // znode containing the lock for the tables
+ public String tableLockZNode;
// Certain ZooKeeper nodes need to be world-readable
public static final ArrayList<ACL> CREATOR_ALL_AND_WORLD_READABLE =
@@ -117,23 +119,23 @@ public class ZooKeeperWatcher implements
/**
* Instantiate a ZooKeeper connection and watcher.
- * @param descriptor Descriptive string that is added to zookeeper sessionid
- * and used as identifier for this instance.
+ * @param identifier string that is passed to RecoverableZookeeper to be used as
+ * identifier for this instance. Use null for default.
* @throws IOException
* @throws ZooKeeperConnectionException
*/
- public ZooKeeperWatcher(Configuration conf, String descriptor,
+ public ZooKeeperWatcher(Configuration conf, String identifier,
Abortable abortable) throws ZooKeeperConnectionException, IOException {
- this(conf, descriptor, abortable, false);
+ this(conf, identifier, abortable, false);
}
/**
* Instantiate a ZooKeeper connection and watcher.
- * @param descriptor Descriptive string that is added to zookeeper sessionid
- * and used as identifier for this instance.
+ * @param identifier string that is passed to RecoverableZookeeper to be used as
+ * identifier for this instance. Use null for default.
* @throws IOException
* @throws ZooKeeperConnectionException
*/
- public ZooKeeperWatcher(Configuration conf, String descriptor,
+ public ZooKeeperWatcher(Configuration conf, String identifier,
Abortable abortable, boolean canCreateBaseZNode)
throws IOException, ZooKeeperConnectionException {
this.conf = conf;
@@ -147,10 +149,10 @@ public class ZooKeeperWatcher implements
this.quorum = ZKConfig.getZKQuorumServersString(conf);
// Identifier will get the sessionid appended later below down when we
// handle the syncconnect event.
- this.identifier = descriptor;
+ this.identifier = identifier;
this.abortable = abortable;
setNodeNames(conf);
- this.recoverableZooKeeper = ZKUtil.connect(conf, quorum, this, descriptor);
+ this.recoverableZooKeeper = ZKUtil.connect(conf, quorum, this, identifier);
if (canCreateBaseZNode) {
createBaseZNodes();
}
@@ -166,6 +168,7 @@ public class ZooKeeperWatcher implements
ZKUtil.createAndFailSilent(this, tableZNode);
ZKUtil.createAndFailSilent(this, splitLogZNode);
ZKUtil.createAndFailSilent(this, backupMasterAddressesZNode);
+ ZKUtil.createAndFailSilent(this, tableLockZNode);
} catch (KeeperException e) {
throw new ZooKeeperConnectionException(
prefix("Unexpected KeeperException creating base node"), e);
@@ -215,6 +218,8 @@ public class ZooKeeperWatcher implements
conf.get("zookeeper.znode.splitlog", HConstants.SPLIT_LOGDIR_NAME));
balancerZNode = ZKUtil.joinZNode(baseZNode,
conf.get("zookeeper.znode.balancer", "balancer"));
+ tableLockZNode = ZKUtil.joinZNode(baseZNode,
+ conf.get("zookeeper.znode.tableLock", "table-lock"));
}
/**
@@ -234,6 +239,10 @@ public class ZooKeeperWatcher implements
listeners.add(0, listener);
}
+ public void unregisterListener(ZooKeeperListener listener) {
+ listeners.remove(listener);
+ }
+
/**
* Get the connection to ZooKeeper.
* @return connection reference to zookeeper
@@ -355,10 +364,10 @@ public class ZooKeeperWatcher implements
if (this.abortable != null) this.abortable.abort(msg,
new KeeperException.SessionExpiredException());
break;
-
+
case ConnectedReadOnly:
- break;
-
+ break;
+
default:
throw new IllegalStateException("Received event is not valid.");
}
@@ -437,7 +446,7 @@ public class ZooKeeperWatcher implements
public void abort(String why, Throwable e) {
this.abortable.abort(why, e);
}
-
+
@Override
public boolean isAborted() {
return this.abortable.isAborted();
@@ -449,4 +458,5 @@ public class ZooKeeperWatcher implements
public String getMasterAddressZNode() {
return this.masterAddressZNode;
}
+
}
Added: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessLockBase.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessLockBase.java?rev=1448867&view=auto
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessLockBase.java (added)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessLockBase.java Fri Feb 22 00:15:52 2013
@@ -0,0 +1,346 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.zookeeper.lock;
+
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.InterProcessLock;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.zookeeper.DeletionListener;
+import org.apache.hadoop.hbase.zookeeper.ZKUtil;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.BadVersionException;
+import org.apache.zookeeper.data.Stat;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * ZooKeeper based HLock implementation. Based on the Shared Locks recipe.
+ * (see:
+ * <a href="http://zookeeper.apache.org/doc/trunk/recipes.html">
+ * ZooKeeper Recipes and Solutions
+ * </a>)
+ */
+@InterfaceAudience.Private
+public abstract class ZKInterProcessLockBase implements InterProcessLock {
+
+ private static final Log LOG = LogFactory.getLog(ZKInterProcessLockBase.class);
+
+ /** ZNode prefix used by processes acquiring reader locks */
+ protected static final String READ_LOCK_CHILD_NODE_PREFIX = "read-";
+
+ /** ZNode prefix used by processes acquiring writer locks */
+ protected static final String WRITE_LOCK_CHILD_NODE_PREFIX = "write-";
+
+ protected final ZooKeeperWatcher zkWatcher;
+ protected final String parentLockNode;
+ protected final String fullyQualifiedZNode;
+ protected final byte[] metadata;
+ protected final MetadataHandler handler;
+
+ // If we acquire a lock, update this field
+ protected final AtomicReference<AcquiredLock> acquiredLock =
+ new AtomicReference<AcquiredLock>(null);
+
+ /**
+ * Represents information about a lock held by this thread.
+ */
+ protected static class AcquiredLock {
+ private final String path;
+ private final int version;
+
+ /**
+ * Store information about a lock.
+ * @param path The path to a lock's ZNode
+ * @param version The current version of the lock's ZNode
+ */
+ public AcquiredLock(String path, int version) {
+ this.path = path;
+ this.version = version;
+ }
+
+ public String getPath() {
+ return path;
+ }
+
+ public int getVersion() {
+ return version;
+ }
+
+ @Override
+ public String toString() {
+ return "AcquiredLockInfo{" +
+ "path='" + path + '\'' +
+ ", version=" + version +
+ '}';
+ }
+ }
+
+ protected static class ZNodeComparator implements Comparator<String> {
+
+ public static final ZNodeComparator COMPARATOR = new ZNodeComparator();
+
+ private ZNodeComparator() {
+ }
+
+ /** Parses sequenceId from the znode name. Zookeeper documentation
+ * states: The sequence number is always fixed length of 10 digits, 0 padded
+ */
+ public static int getChildSequenceId(String childZNode) {
+ Preconditions.checkNotNull(childZNode);
+ assert childZNode.length() >= 10;
+ String sequenceIdStr = childZNode.substring(childZNode.length() - 10);
+ return Integer.parseInt(sequenceIdStr);
+ }
+
+ @Override
+ public int compare(String zNode1, String zNode2) {
+ int seq1 = getChildSequenceId(zNode1);
+ int seq2 = getChildSequenceId(zNode2);
+ return seq1 - seq2;
+ }
+ }
+
+ /**
+ * Called by implementing classes.
+ * @param zkWatcher
+ * @param parentLockNode The lock ZNode path
+ * @param metadata
+ * @param handler
+ * @param childNode The prefix for child nodes created under the parent
+ */
+ protected ZKInterProcessLockBase(ZooKeeperWatcher zkWatcher,
+ String parentLockNode, byte[] metadata, MetadataHandler handler, String childNode) {
+ this.zkWatcher = zkWatcher;
+ this.parentLockNode = parentLockNode;
+ this.fullyQualifiedZNode = ZKUtil.joinZNode(parentLockNode, childNode);
+ this.metadata = metadata;
+ this.handler = handler;
+ try {
+ ZKUtil.createWithParents(zkWatcher, parentLockNode);
+ } catch (KeeperException ex) {
+ LOG.warn("Failed to create znode:" + parentLockNode, ex);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void acquire() throws IOException, InterruptedException {
+ tryAcquire(-1);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean tryAcquire(long timeoutMs)
+ throws IOException, InterruptedException {
+ boolean hasTimeout = timeoutMs != -1;
+ long waitUntilMs =
+ hasTimeout ?EnvironmentEdgeManager.currentTimeMillis() + timeoutMs : -1;
+ String createdZNode = createLockZNode();
+ while (true) {
+ List<String> children;
+ try {
+ children = ZKUtil.listChildrenNoWatch(zkWatcher, parentLockNode);
+ } catch (KeeperException e) {
+ LOG.error("Unexpected ZooKeeper error when listing children", e);
+ throw new IOException("Unexpected ZooKeeper exception", e);
+ }
+ String pathToWatch;
+ if ((pathToWatch = getLockPath(createdZNode, children)) == null) {
+ break;
+ }
+ CountDownLatch deletedLatch = new CountDownLatch(1);
+ String zkPathToWatch =
+ ZKUtil.joinZNode(parentLockNode, pathToWatch);
+ DeletionListener deletionListener =
+ new DeletionListener(zkWatcher, zkPathToWatch, deletedLatch);
+ zkWatcher.registerListener(deletionListener);
+ try {
+ if (ZKUtil.setWatchIfNodeExists(zkWatcher, zkPathToWatch)) {
+ // Wait for the watcher to fire
+ if (hasTimeout) {
+ long remainingMs = waitUntilMs - EnvironmentEdgeManager.currentTimeMillis();
+ if (remainingMs < 0 ||
+ !deletedLatch.await(remainingMs, TimeUnit.MILLISECONDS)) {
+ LOG.warn("Unable to acquire the lock in " + timeoutMs +
+ " milliseconds.");
+ try {
+ ZKUtil.deleteNode(zkWatcher, createdZNode);
+ } catch (KeeperException e) {
+ LOG.warn("Unable to remove ZNode " + createdZNode);
+ }
+ return false;
+ }
+ } else {
+ deletedLatch.await();
+ }
+ if (deletionListener.hasException()) {
+ Throwable t = deletionListener.getException();
+ throw new IOException("Exception in the watcher", t);
+ }
+ }
+ } catch (KeeperException e) {
+ throw new IOException("Unexpected ZooKeeper exception", e);
+ } finally {
+ zkWatcher.unregisterListener(deletionListener);
+ }
+ }
+ updateAcquiredLock(createdZNode);
+ LOG.debug("Successfully acquired a lock for " + createdZNode);
+ return true;
+ }
+
+ private String createLockZNode() {
+ try {
+ return ZKUtil.createNodeIfNotExistsNoWatch(zkWatcher, fullyQualifiedZNode,
+ metadata, CreateMode.EPHEMERAL_SEQUENTIAL);
+ } catch (KeeperException ex) {
+ LOG.warn("Failed to create znode: " + fullyQualifiedZNode, ex);
+ return null;
+ }
+ }
+
+ /**
+ * Check if a child znode represents a write lock.
+ * @param child The child znode we want to check.
+ * @return whether the child znode represents a write lock
+ */
+ protected static boolean isChildWriteLock(String child) {
+ int idx = child.lastIndexOf(ZKUtil.ZNODE_PATH_SEPARATOR);
+ String suffix = child.substring(idx + 1);
+ return suffix.startsWith(WRITE_LOCK_CHILD_NODE_PREFIX);
+ }
+
+ /**
+ * Update state as to indicate that a lock is held
+ * @param createdZNode The lock znode
+ * @throws IOException If an unrecoverable ZooKeeper error occurs
+ */
+ protected void updateAcquiredLock(String createdZNode) throws IOException {
+ Stat stat = new Stat();
+ byte[] data = null;
+ Exception ex = null;
+ try {
+ data = ZKUtil.getDataNoWatch(zkWatcher, createdZNode, stat);
+ } catch (KeeperException e) {
+ LOG.warn("Cannot getData for znode:" + createdZNode, e);
+ ex = e;
+ }
+ if (data == null) {
+ LOG.error("Can't acquire a lock on a non-existent node " + createdZNode);
+ throw new IllegalStateException("ZNode " + createdZNode +
+ "no longer exists!", ex);
+ }
+ AcquiredLock newLock = new AcquiredLock(createdZNode, stat.getVersion());
+ if (!acquiredLock.compareAndSet(null, newLock)) {
+ LOG.error("The lock " + fullyQualifiedZNode +
+ " has already been acquired by another process!");
+ throw new IllegalStateException(fullyQualifiedZNode +
+ " is held by another process");
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void release() throws IOException, InterruptedException {
+ AcquiredLock lock = acquiredLock.get();
+ if (lock == null) {
+ LOG.error("Cannot release lock" +
+ ", process does not have a lock for " + fullyQualifiedZNode);
+ throw new IllegalStateException("No lock held for " + fullyQualifiedZNode);
+ }
+ try {
+ if (ZKUtil.checkExists(zkWatcher, lock.getPath()) != -1) {
+ ZKUtil.deleteNode(zkWatcher, lock.getPath(), lock.getVersion());
+ if (!acquiredLock.compareAndSet(lock, null)) {
+ LOG.debug("Current process no longer holds " + lock + " for " +
+ fullyQualifiedZNode);
+ throw new IllegalStateException("Not holding a lock for " +
+ fullyQualifiedZNode +"!");
+ }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Successfully released " + lock.getPath());
+ }
+ } catch (BadVersionException e) {
+ throw new IllegalStateException(e);
+ } catch (KeeperException e) {
+ throw new IOException(e);
+ }
+ }
+
+ /**
+ * Process metadata stored in a ZNode using a callback object passed to
+ * this instance.
+ * <p>
+ * @param lockZNode The node holding the metadata
+ * @return True if metadata was ready and processed
+ * @throws IOException If an unexpected ZooKeeper error occurs
+ * @throws InterruptedException If interrupted when reading the metadata
+ */
+ protected boolean handleLockMetadata(String lockZNode)
+ throws IOException, InterruptedException {
+ byte[] metadata = null;
+ try {
+ metadata = ZKUtil.getData(zkWatcher, lockZNode);
+ } catch (KeeperException ex) {
+ LOG.warn("Cannot getData for znode:" + lockZNode, ex);
+ }
+ if (metadata == null) {
+ return false;
+ }
+ if (handler != null) {
+ handler.handleMetadata(metadata);
+ }
+ return true;
+ }
+
+ /**
+ * Determine based on a list of children under a ZNode, whether or not a
+ * process which created a specified ZNode has obtained a lock. If a lock is
+ * not obtained, return the path that we should watch awaiting its deletion.
+ * Otherwise, return null.
+ * This method is abstract as the logic for determining whether or not a
+ * lock is obtained depends on the type of lock being implemented.
+ * @param myZNode The ZNode created by the process attempting to acquire
+ * a lock
+ * @param children List of all child ZNodes under the lock's parent ZNode
+ * @return The path to watch, or null if myZNode can represent a correctly
+ * acquired lock.
+ */
+ protected abstract String getLockPath(String myZNode, List<String> children)
+ throws IOException, InterruptedException;
+}
Added: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadLock.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadLock.java?rev=1448867&view=auto
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadLock.java (added)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadLock.java Fri Feb 22 00:15:52 2013
@@ -0,0 +1,83 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.zookeeper.lock;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.zookeeper.ZKUtil;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+
+/**
+ * ZooKeeper based read lock: does not exclude other read locks, but excludes
+ * and is excluded by write locks.
+ */
+@InterfaceAudience.Private
+public class ZKInterProcessReadLock extends ZKInterProcessLockBase {
+
+ private static final Log LOG = LogFactory.getLog(ZKInterProcessReadLock.class);
+
+ public ZKInterProcessReadLock(ZooKeeperWatcher zooKeeperWatcher,
+ String znode, byte[] metadata, MetadataHandler handler) {
+ super(zooKeeperWatcher, znode, metadata, handler, READ_LOCK_CHILD_NODE_PREFIX);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected String getLockPath(String createdZNode, List<String> children)
+ throws IOException, InterruptedException {
+ TreeSet<String> writeChildren =
+ new TreeSet<String>(ZNodeComparator.COMPARATOR);
+ for (String child : children) {
+ if (isChildWriteLock(child)) {
+ writeChildren.add(child);
+ }
+ }
+ if (writeChildren.isEmpty()) {
+ return null;
+ }
+ SortedSet<String> lowerChildren = writeChildren.headSet(createdZNode);
+ if (lowerChildren.isEmpty()) {
+ return null;
+ }
+ String pathToWatch = lowerChildren.last();
+ String nodeHoldingLock = lowerChildren.first();
+ String znode = ZKUtil.joinZNode(parentLockNode, nodeHoldingLock);
+ try {
+ handleLockMetadata(znode);
+ } catch (IOException e) {
+ LOG.warn("Error processing lock metadata in " + nodeHoldingLock, e);
+ }
+ return pathToWatch;
+ }
+
+ @Override
+ public void reapAllLocks() throws IOException {
+ throw new UnsupportedOperationException(
+ "Lock reaping is not supported for ZK based read locks");
+ }
+}
Added: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadWriteLock.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadWriteLock.java?rev=1448867&view=auto
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadWriteLock.java (added)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessReadWriteLock.java Fri Feb 22 00:15:52 2013
@@ -0,0 +1,66 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.zookeeper.lock;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.InterProcessLock.MetadataHandler;
+import org.apache.hadoop.hbase.InterProcessReadWriteLock;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+
+/**
+ * ZooKeeper based implementation of {@link InterProcessReadWriteLock}. This lock is fair,
+ * not reentrant, and not revocable.
+ */
+@InterfaceAudience.Private
+public class ZKInterProcessReadWriteLock implements InterProcessReadWriteLock {
+
+ private final ZooKeeperWatcher zkWatcher;
+ private final String znode;
+ private final MetadataHandler handler;
+
+ /**
+ * Creates a DistributedReadWriteLock instance.
+ * @param zkWatcher
+ * @param znode ZNode path for the lock
+ * @param handler An object that will handle de-serializing and processing
+ * the metadata associated with reader or writer locks
+ * created by this object or null if none desired.
+ */
+ public ZKInterProcessReadWriteLock(ZooKeeperWatcher zkWatcher, String znode,
+ MetadataHandler handler) {
+ this.zkWatcher = zkWatcher;
+ this.znode = znode;
+ this.handler = handler;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public ZKInterProcessReadLock readLock(byte[] metadata) {
+ return new ZKInterProcessReadLock(zkWatcher, znode, metadata, handler);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public ZKInterProcessWriteLock writeLock(byte[] metadata) {
+ return new ZKInterProcessWriteLock(zkWatcher, znode, metadata, handler);
+ }
+}
Added: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessWriteLock.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessWriteLock.java?rev=1448867&view=auto
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessWriteLock.java (added)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/lock/ZKInterProcessWriteLock.java Fri Feb 22 00:15:52 2013
@@ -0,0 +1,98 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.zookeeper.lock;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.TreeSet;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.zookeeper.ZKUtil;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * ZooKeeper based write lock:
+ */
+@InterfaceAudience.Private
+public class ZKInterProcessWriteLock extends ZKInterProcessLockBase {
+
+ private static final Log LOG = LogFactory.getLog(ZKInterProcessWriteLock.class);
+
+ public ZKInterProcessWriteLock(ZooKeeperWatcher zooKeeperWatcher,
+ String znode, byte[] metadata, MetadataHandler handler) {
+ super(zooKeeperWatcher, znode, metadata, handler, WRITE_LOCK_CHILD_NODE_PREFIX);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected String getLockPath(String createdZNode, List<String> children)
+ throws IOException, InterruptedException {
+ TreeSet<String> sortedChildren =
+ new TreeSet<String>(ZNodeComparator.COMPARATOR);
+ sortedChildren.addAll(children);
+ String pathToWatch = sortedChildren.lower(createdZNode);
+ if (pathToWatch != null) {
+ String nodeHoldingLock = sortedChildren.first();
+ String znode = ZKUtil.joinZNode(parentLockNode, nodeHoldingLock);
+ try {
+ handleLockMetadata(znode);
+ } catch (IOException e) {
+ LOG.warn("Error processing lock metadata in " + nodeHoldingLock, e);
+ }
+ }
+ return pathToWatch;
+ }
+
+ /**
+ * Referred in zk recipe as "Revocable Shared Locks with Freaking Laser Beams"
+ * (http://zookeeper.apache.org/doc/trunk/recipes.html).
+ */
+ public void reapAllLocks() throws IOException {
+ List<String> children;
+ try {
+ children = ZKUtil.listChildrenNoWatch(zkWatcher, parentLockNode);
+ } catch (KeeperException e) {
+ LOG.error("Unexpected ZooKeeper error when listing children", e);
+ throw new IOException("Unexpected ZooKeeper exception", e);
+ }
+
+ KeeperException deferred = null;
+ for (String child : children) {
+ if (isChildWriteLock(child)) {
+ String znode = ZKUtil.joinZNode(parentLockNode, child);
+ LOG.info("Reaping write lock for znode:" + znode);
+ try {
+ ZKUtil.deleteNodeFailSilent(zkWatcher, znode);
+ } catch (KeeperException ex) {
+ LOG.warn("Error reaping the znode for write lock :" + znode);
+ deferred = ex;
+ }
+ }
+ }
+ if (deferred != null) {
+ throw new IOException("ZK exception while reaping locks:", deferred);
+ }
+ }
+}
Modified: hbase/trunk/hbase-server/src/main/resources/hbase-default.xml
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/resources/hbase-default.xml?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/resources/hbase-default.xml (original)
+++ hbase/trunk/hbase-server/src/main/resources/hbase-default.xml Fri Feb 22 00:15:52 2013
@@ -837,6 +837,15 @@
</description>
</property>
<property>
+ <name>hbase.table.lock.enable</name>
+ <value>true</value>
+ <description>
+ Set to true to enable locking the table in zookeeper for schema change operations.
+ Table locking from master prevents concurrent schema modifications to corrupt table
+ state.
+ </description>
+ </property>
+ <property>
<name>dfs.support.append</name>
<value>true</value>
<description>Does HDFS allow appends to files?
Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/MultithreadedTestUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/MultithreadedTestUtil.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/MultithreadedTestUtil.java (original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/MultithreadedTestUtil.java Fri Feb 22 00:15:52 2013
@@ -18,8 +18,11 @@
*/
package org.apache.hadoop.hbase;
-import java.util.Set;
import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -91,7 +94,7 @@ public abstract class MultithreadedTestU
stopped = s;
}
}
-
+
public void stop() throws Exception {
synchronized (this) {
stopped = true;
@@ -130,7 +133,7 @@ public abstract class MultithreadedTestU
this.stopped = true;
}
}
-
+
/**
* A test thread that performs a repeating operation.
*/
@@ -138,13 +141,48 @@ public abstract class MultithreadedTestU
public RepeatingTestThread(TestContext ctx) {
super(ctx);
}
-
+
public final void doWork() throws Exception {
while (ctx.shouldRun() && !stopped) {
doAnAction();
}
}
-
+
public abstract void doAnAction() throws Exception;
}
+
+ /**
+ * Verify that no assertions have failed inside a future.
+ * Used for unit tests that spawn threads. E.g.,
+ * <p>
+ * <code>
+ * List<Future<Void>> results = Lists.newArrayList();
+ * Future<Void> f = executor.submit(new Callable<Void> {
+ * public Void call() {
+ * assertTrue(someMethod());
+ * }
+ * });
+ * results.add(f);
+ * assertOnFutures(results);
+ * </code>
+ * @param threadResults A list of futures
+ * @param <T>
+ * @throws InterruptedException If interrupted when waiting for a result
+ * from one of the futures
+ * @throws ExecutionException If an exception other than AssertionError
+ * occurs inside any of the futures
+ */
+ public static <T> void assertOnFutures(List<Future<T>> threadResults)
+ throws InterruptedException, ExecutionException {
+ for (Future<T> threadResult : threadResults) {
+ try {
+ threadResult.get();
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof AssertionError) {
+ throw (AssertionError) e.getCause();
+ }
+ throw e;
+ }
+ }
+ }
}
Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestAssignmentManager.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestAssignmentManager.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestAssignmentManager.java (original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestAssignmentManager.java Fri Feb 22 00:15:52 2013
@@ -52,6 +52,7 @@ import org.apache.hadoop.hbase.executor.
import org.apache.hadoop.hbase.executor.ExecutorService;
import org.apache.hadoop.hbase.executor.ExecutorService.ExecutorType;
import org.apache.hadoop.hbase.master.RegionState.State;
+import org.apache.hadoop.hbase.master.TableLockManager.NullTableLockManager;
import org.apache.hadoop.hbase.master.balancer.DefaultLoadBalancer;
import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory;
import org.apache.hadoop.hbase.master.handler.EnableTableHandler;
@@ -182,7 +183,7 @@ public class TestAssignmentManager {
* @throws IOException
* @throws KeeperException
* @throws InterruptedException
- * @throws DeserializationException
+ * @throws DeserializationException
*/
@Test(timeout = 5000)
public void testBalanceOnMasterFailoverScenarioWithOpenedNode()
@@ -346,7 +347,7 @@ public class TestAssignmentManager {
* from one server to another mocking regionserver responding over zk.
* @throws IOException
* @throws KeeperException
- * @throws DeserializationException
+ * @throws DeserializationException
*/
@Test
public void testBalance()
@@ -361,7 +362,7 @@ public class TestAssignmentManager {
.getConfiguration());
// Create an AM.
AssignmentManager am = new AssignmentManager(this.server,
- this.serverManager, ct, balancer, executor, null);
+ this.serverManager, ct, balancer, executor, null, master.getTableLockManager());
am.failoverCleanupDone.set(true);
try {
// Make sure our new AM gets callbacks; once registered, can't unregister.
@@ -442,7 +443,7 @@ public class TestAssignmentManager {
* To test closed region handler to remove rit and delete corresponding znode
* if region in pending close or closing while processing shutdown of a region
* server.(HBASE-5927).
- *
+ *
* @throws KeeperException
* @throws IOException
* @throws ServiceException
@@ -453,12 +454,12 @@ public class TestAssignmentManager {
testCaseWithPartiallyDisabledState(Table.State.DISABLING);
testCaseWithPartiallyDisabledState(Table.State.DISABLED);
}
-
-
+
+
/**
* To test if the split region is removed from RIT if the region was in SPLITTING state but the RS
* has actually completed the splitting in META but went down. See HBASE-6070 and also HBASE-5806
- *
+ *
* @throws KeeperException
* @throws IOException
*/
@@ -469,7 +470,7 @@ public class TestAssignmentManager {
// false indicate the region is not split
testCaseWithSplitRegionPartial(false);
}
-
+
private void testCaseWithSplitRegionPartial(boolean regionSplitDone) throws KeeperException,
IOException, NodeExistsException, InterruptedException, ServiceException {
// Create and startup an executor. This is used by AssignmentManager
@@ -530,7 +531,7 @@ public class TestAssignmentManager {
// Create an AM.
AssignmentManager am = new AssignmentManager(this.server,
- this.serverManager, ct, balancer, executor, null);
+ this.serverManager, ct, balancer, executor, null, master.getTableLockManager());
// adding region to regions and servers maps.
am.regionOnline(REGIONINFO, SERVERNAME_A);
// adding region in pending close.
@@ -651,7 +652,7 @@ public class TestAssignmentManager {
.getConfiguration());
// Create an AM.
AssignmentManager am = new AssignmentManager(this.server,
- this.serverManager, ct, balancer, null, null);
+ this.serverManager, ct, balancer, null, null, master.getTableLockManager());
try {
// First make sure my mock up basically works. Unassign a region.
unassign(am, SERVERNAME_A, hri);
@@ -679,7 +680,7 @@ public class TestAssignmentManager {
* Tests the processDeadServersAndRegionsInTransition should not fail with NPE
* when it failed to get the children. Let's abort the system in this
* situation
- * @throws ServiceException
+ * @throws ServiceException
*/
@Test(timeout = 5000)
public void testProcessDeadServersAndRegionsInTransitionShouldNotFailWithNPE()
@@ -708,7 +709,7 @@ public class TestAssignmentManager {
}
}
/**
- * TestCase verifies that the regionPlan is updated whenever a region fails to open
+ * TestCase verifies that the regionPlan is updated whenever a region fails to open
* and the master tries to process RS_ZK_FAILED_OPEN state.(HBASE-5546).
*/
@Test(timeout = 5000)
@@ -795,7 +796,7 @@ public class TestAssignmentManager {
this.gate.set(true);
return randomServerName;
}
-
+
@Override
public Map<ServerName, List<HRegionInfo>> retainAssignment(
Map<HRegionInfo, ServerName> regions, List<ServerName> servers) {
@@ -836,7 +837,7 @@ public class TestAssignmentManager {
/**
* Test verifies whether assignment is skipped for regions of tables in DISABLING state during
* clean cluster startup. See HBASE-6281.
- *
+ *
* @throws KeeperException
* @throws IOException
* @throws Exception
@@ -882,7 +883,7 @@ public class TestAssignmentManager {
/**
* Test verifies whether all the enabling table regions assigned only once during master startup.
- *
+ *
* @throws KeeperException
* @throws IOException
* @throws Exception
@@ -902,7 +903,8 @@ public class TestAssignmentManager {
try {
// set table in enabling state.
am.getZKTable().setEnablingTable(REGIONINFO.getTableNameAsString());
- new EnableTableHandler(server, REGIONINFO.getTableName(), am.getCatalogTracker(), am, true)
+ new EnableTableHandler(server, REGIONINFO.getTableName(), am.getCatalogTracker(),
+ am, new NullTableLockManager(), true).prepare()
.process();
assertEquals("Number of assignments should be 1.", 1, assignmentCount);
assertTrue("Table should be enabled.",
@@ -1071,7 +1073,7 @@ public class TestAssignmentManager {
ExecutorService executor = startupMasterExecutor("mockedAMExecutor");
this.balancer = LoadBalancerFactory.getLoadBalancer(server.getConfiguration());
AssignmentManagerWithExtrasForTesting am = new AssignmentManagerWithExtrasForTesting(
- server, manager, ct, this.balancer, executor);
+ server, manager, ct, this.balancer, executor, new NullTableLockManager());
return am;
}
@@ -1090,8 +1092,9 @@ public class TestAssignmentManager {
public AssignmentManagerWithExtrasForTesting(
final Server master, final ServerManager serverManager,
final CatalogTracker catalogTracker, final LoadBalancer balancer,
- final ExecutorService service) throws KeeperException, IOException {
- super(master, serverManager, catalogTracker, balancer, service, null);
+ final ExecutorService service, final TableLockManager tableLockManager)
+ throws KeeperException, IOException {
+ super(master, serverManager, catalogTracker, balancer, service, null, tableLockManager);
this.es = service;
this.ct = catalogTracker;
}
Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestCatalogJanitor.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestCatalogJanitor.java?rev=1448867&r1=1448866&r2=1448867&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestCatalogJanitor.java (original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestCatalogJanitor.java Fri Feb 22 00:15:52 2013
@@ -320,6 +320,11 @@ public class TestCatalogJanitor {
@Override
public void deleteColumn(byte[] tableName, byte[] columnName) throws IOException { }
+
+ @Override
+ public TableLockManager getTableLockManager() {
+ return null;
+ }
}
@Test