diff --git a/CHANGES.txt b/CHANGES.txt
index faa7e58465f..fe83fc650d4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -231,6 +231,8 @@ Release 0.91.0 - Unreleased
    HBASE-4310  SlabCache metrics bugfix (Li Pi)
    HBASE-4283  HBaseAdmin never recovers from restarted cluster (Lars Hofhansl)
    HBASE-4315  RPC logging too verbose (todd)
+   HBASE-4273  java.lang.NullPointerException when a table is being disabled and
+               HMaster restarts (Ming Ma)
 
   IMPROVEMENTS
    HBASE-3290  Max Compaction Size (Nicolas Spiegelberg via Stack)  
diff --git a/src/main/java/org/apache/hadoop/hbase/TableNotEnabledException.java b/src/main/java/org/apache/hadoop/hbase/TableNotEnabledException.java
new file mode 100644
index 00000000000..d0392d52f6e
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/hbase/TableNotEnabledException.java
@@ -0,0 +1,50 @@
+/**
+ * Copyright 2011 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;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Thrown if a table should be enabled but is not
+ */
+public class TableNotEnabledException extends IOException {
+  private static final long serialVersionUID = 262144L;
+  /** default constructor */
+  public TableNotEnabledException() {
+    super();
+  }
+
+  /**
+   * Constructor
+   * @param s message
+   */
+  public TableNotEnabledException(String s) {
+    super(s);
+  }
+
+  /**
+   * @param tableName Name of table that is not enabled
+   */
+  public TableNotEnabledException(byte[] tableName) {
+    this(Bytes.toString(tableName));
+  }
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java b/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
index 88d72a02a1e..f17036e1eeb 100644
--- a/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
+++ b/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
@@ -565,8 +565,12 @@ public class HBaseAdmin implements Abortable, Closeable {
   /**
    * Enable a table.  May timeout.  Use {@link #enableTableAsync(byte[])}
    * and {@link #isTableEnabled(byte[])} instead.
+   * The table has to be in disabled state for it to be enabled.
    * @param tableName name of the table
    * @throws IOException if a remote or network exception occurs
+   * There could be couple types of IOException
+   * TableNotFoundException means the table doesn't exist.
+   * TableNotDisabledException means the table isn't in disabled state.
    * @see #isTableEnabled(byte[])
    * @see #disableTable(byte[])
    * @see #enableTableAsync(byte[])
@@ -706,8 +710,12 @@ public class HBaseAdmin implements Abortable, Closeable {
    * Disable table and wait on completion.  May timeout eventually.  Use
    * {@link #disableTableAsync(byte[])} and {@link #isTableDisabled(String)}
    * instead.
+   * The table has to be in enabled state for it to be disabled.
    * @param tableName
    * @throws IOException
+   * There could be couple types of IOException
+   * TableNotFoundException means the table doesn't exist.
+   * TableNotEnabledException means the table isn't in enabled state.
    */
   public void disableTable(final byte [] tableName)
   throws IOException {
diff --git a/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java b/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java
index 13af86a035c..698c7d3518c 100644
--- a/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java
+++ b/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java
@@ -1830,12 +1830,20 @@ public class AssignmentManager extends ZooKeeperListener {
       ServerName regionLocation = region.getSecond();
       String tableName = regionInfo.getTableNameAsString();
       if (regionLocation == null) {
-        // Region not being served, add to region map with no assignment
-        // If this needs to be assigned out, it will also be in ZK as RIT
-        // add if the table is not in disabled and enabling state
-        if (false == checkIfRegionBelongsToDisabled(regionInfo)
-            && false == checkIfRegionsBelongsToEnabling(regionInfo)) {
-          regions.put(regionInfo, regionLocation);
+        // regionLocation could be null if createTable didn't finish properly.
+        // When createTable is in progress, HMaster restarts.
+        // Some regions have been added to .META., but have not been assigned.
+        // When this happens, the region's table must be in ENABLING state.
+        // It can't be in ENABLED state as that is set when all regions are
+        // assigned.
+        // It can't be in DISABLING state, because DISABLING state transitions
+        // from ENABLED state when application calls disableTable.
+        // It can't be in DISABLED state, because DISABLED states transitions
+        // from DISABLING state.
+        if (false == checkIfRegionsBelongsToEnabling(regionInfo)) {
+          LOG.warn("Region " + regionInfo.getEncodedName() +
+            " has null regionLocation." + " But its table " + tableName +
+            " isn't in ENABLING state.");
         }
         addTheTablesInPartialState(disablingTables, enablingTables, regionInfo,
             tableName);
@@ -1901,7 +1909,7 @@ public class AssignmentManager extends ZooKeeperListener {
             + " is in DISABLING state.  Hence recovering by moving the table"
             + " to DISABLED state.");
         new DisableTableHandler(this.master, tableName.getBytes(),
-            catalogTracker, this).process();
+            catalogTracker, this, true).process();
       }
     }
     return isWatcherCreated;
@@ -1931,7 +1939,7 @@ public class AssignmentManager extends ZooKeeperListener {
             + " is in ENABLING state.  Hence recovering by moving the table"
             + " to ENABLED state.");
         new EnableTableHandler(this.master, tableName.getBytes(),
-            catalogTracker, this).process();
+            catalogTracker, this, true).process();
       }
     }
   }
diff --git a/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
index 6529a6fd3fe..de93dc546aa 100644
--- a/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
+++ b/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
@@ -1029,7 +1029,7 @@ implements HMasterInterface, HMasterRegionInterface, MasterServices, Server {
       cpHost.preEnableTable(tableName);
     }
     this.executorService.submit(new EnableTableHandler(this, tableName,
-      catalogTracker, assignmentManager));
+      catalogTracker, assignmentManager, false));
     if (cpHost != null) {
       cpHost.postEnableTable(tableName);
     }
@@ -1040,7 +1040,7 @@ implements HMasterInterface, HMasterRegionInterface, MasterServices, Server {
       cpHost.preDisableTable(tableName);
     }
     this.executorService.submit(new DisableTableHandler(this, tableName,
-      catalogTracker, assignmentManager));
+      catalogTracker, assignmentManager, false));
     if (cpHost != null) {
       cpHost.postDisableTable(tableName);
     }
diff --git a/src/main/java/org/apache/hadoop/hbase/master/handler/DisableTableHandler.java b/src/main/java/org/apache/hadoop/hbase/master/handler/DisableTableHandler.java
index 6b2a616de23..5af0690207e 100644
--- a/src/main/java/org/apache/hadoop/hbase/master/handler/DisableTableHandler.java
+++ b/src/main/java/org/apache/hadoop/hbase/master/handler/DisableTableHandler.java
@@ -27,6 +27,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.Server;
+import org.apache.hadoop.hbase.TableNotEnabledException;
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.catalog.CatalogTracker;
 import org.apache.hadoop.hbase.catalog.MetaReader;
@@ -46,8 +47,9 @@ public class DisableTableHandler extends EventHandler {
   private final AssignmentManager assignmentManager;
 
   public DisableTableHandler(Server server, byte [] tableName,
-      CatalogTracker catalogTracker, AssignmentManager assignmentManager)
-  throws TableNotFoundException, IOException {
+      CatalogTracker catalogTracker, AssignmentManager assignmentManager,
+      boolean skipTableStateCheck)
+  throws TableNotFoundException, TableNotEnabledException, IOException {
     super(server, EventType.C_M_DISABLE_TABLE);
     this.tableName = tableName;
     this.tableNameStr = Bytes.toString(this.tableName);
@@ -57,7 +59,25 @@ public class DisableTableHandler extends EventHandler {
     //       part of old master rewrite, schema to zk to check for table
     //       existence and such
     if (!MetaReader.tableExists(catalogTracker, this.tableNameStr)) {
-      throw new TableNotFoundException(Bytes.toString(tableName));
+      throw new TableNotFoundException(this.tableNameStr);
+    }
+
+    // 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().checkEnabledAndSetDisablingTable
+          (this.tableNameStr)) {
+          LOG.info("Table " + tableNameStr + " isn't enabled; skipping disable");
+          throw new TableNotEnabledException(this.tableNameStr);
+        }
+      } catch (KeeperException e) {
+        throw new IOException("Unable to ensure that the table will be" +
+          " disabling because of a ZooKeeper issue", e);
+      }
     }
   }
 
@@ -67,9 +87,10 @@ public class DisableTableHandler extends EventHandler {
     if(server != null && server.getServerName() != null) {
       name = server.getServerName().toString();
     }
-    return getClass().getSimpleName() + "-" + name + "-" + getSeqid() + "-" + tableNameStr;
+    return getClass().getSimpleName() + "-" + name + "-" + getSeqid() + "-" +
+      tableNameStr;
   }
-  
+
   @Override
   public void process() {
     try {
@@ -83,18 +104,14 @@ public class DisableTableHandler extends EventHandler {
   }
 
   private void handleDisableTable() throws IOException, KeeperException {
-    if (this.assignmentManager.getZKTable().isDisabledTable(this.tableNameStr)) {
-      LOG.info("Table " + tableNameStr + " already disabled; skipping disable");
-      return;
-    }
     // Set table disabling flag up in zk.
     this.assignmentManager.getZKTable().setDisablingTable(this.tableNameStr);
     boolean done = false;
     while (true) {
       // Get list of online regions that are of this table.  Regions that are
       // already closed will not be included in this list; i.e. the returned
-      // list is not ALL regions in a table, its all online regions according to
-      // the in-memory state on this master.
+      // list is not ALL regions in a table, its all online regions according
+      // to the in-memory state on this master.
       final List<HRegionInfo> regions =
         this.assignmentManager.getRegionsOfTable(tableName);
       if (regions.size() == 0) {
@@ -159,4 +176,4 @@ public class DisableTableHandler extends EventHandler {
       return regions != null && regions.isEmpty();
     }
   }
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java b/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java
index 2380d64f55d..b6dc1c04eb9 100644
--- a/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java
+++ b/src/main/java/org/apache/hadoop/hbase/master/handler/EnableTableHandler.java
@@ -27,6 +27,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.Server;
+import org.apache.hadoop.hbase.TableNotDisabledException;
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.catalog.CatalogTracker;
 import org.apache.hadoop.hbase.catalog.MetaReader;
@@ -47,8 +48,9 @@ public class EnableTableHandler extends EventHandler {
   private final CatalogTracker ct;
 
   public EnableTableHandler(Server server, byte [] tableName,
-      CatalogTracker catalogTracker, AssignmentManager assignmentManager)
-  throws TableNotFoundException, IOException {
+      CatalogTracker catalogTracker, AssignmentManager assignmentManager,
+      boolean skipTableStateCheck)
+  throws TableNotFoundException, TableNotDisabledException, IOException {
     super(server, EventType.C_M_ENABLE_TABLE);
     this.tableName = tableName;
     this.tableNameStr = Bytes.toString(tableName);
@@ -58,6 +60,24 @@ public class EnableTableHandler extends EventHandler {
     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);
+      }
+    }
   }
 
   @Override
@@ -83,10 +103,6 @@ public class EnableTableHandler extends EventHandler {
   }
 
   private void handleEnableTable() throws IOException, KeeperException {
-    if (this.assignmentManager.getZKTable().isEnabledTable(this.tableNameStr)) {
-      LOG.info("Table " + tableNameStr + " is already enabled; skipping enable");
-      return;
-    }
     // I could check table is disabling and if so, not enable but require
     // that user first finish disabling but that might be obnoxious.
 
@@ -121,7 +137,8 @@ public class EnableTableHandler extends EventHandler {
       }
     }
     // Flip the table to enabled.
-    if (done) this.assignmentManager.getZKTable().setEnabledTable(this.tableNameStr);
+    if (done) this.assignmentManager.getZKTable().setEnabledTable(
+      this.tableNameStr);
     LOG.info("Enabled table is done=" + done);
   }
 
@@ -131,7 +148,8 @@ public class EnableTableHandler extends EventHandler {
    * been onlined; i.e. List of regions that need onlining.
    * @throws IOException
    */
-  private List<HRegionInfo> regionsToAssign(final List<HRegionInfo> regionsInMeta)
+  private List<HRegionInfo> regionsToAssign(
+    final List<HRegionInfo> regionsInMeta)
   throws IOException {
     final List<HRegionInfo> onlineRegions =
       this.assignmentManager.getRegionsOfTable(tableName);
@@ -186,4 +204,4 @@ public class EnableTableHandler extends EventHandler {
       return regions != null && regions.size() >= this.countOfRegionsInTable;
     }
   }
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKTable.java b/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKTable.java
index a340332d063..4930f669875 100644
--- a/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKTable.java
+++ b/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKTable.java
@@ -189,6 +189,42 @@ public class ZKTable {
     }
   }
 
+  /**
+   * Sets the specified table as ENABLING in zookeeper atomically
+   * If the table isn't in DISABLED state, no operation is performed
+   * @param tableName
+   * @return if the operation succeeds or not
+   * @throws KeeperException unexpected zookeeper exception
+   */
+  public boolean checkDisabledAndSetEnablingTable(final String tableName)
+    throws KeeperException {
+    synchronized (this.cache) {
+      if (!isDisabledTable(tableName)) {
+        return false;
+      }
+      setTableState(tableName, TableState.ENABLING);
+      return true;
+    }
+  }
+
+  /**
+   * Sets the specified table as DISABLING in zookeeper atomically
+   * If the table isn't in ENABLED state, no operation is performed
+   * @param tableName
+   * @return if the operation succeeds or not
+   * @throws KeeperException unexpected zookeeper exception
+   */
+  public boolean checkEnabledAndSetDisablingTable(final String tableName)
+    throws KeeperException {
+    synchronized (this.cache) {
+      if (!isEnabledTable(tableName)) {
+        return false;
+      }
+      setTableState(tableName, TableState.DISABLING);
+      return true;
+    }
+  }
+
   private void setTableState(final String tableName, final TableState state)
   throws KeeperException {
     String znode = ZKUtil.joinZNode(this.watcher.tableZNode, tableName);
@@ -366,4 +402,4 @@ public class ZKTable {
     }
     return disabledTables;
   }
-}
\ No newline at end of file
+}
diff --git a/src/test/java/org/apache/hadoop/hbase/avro/TestAvroServer.java b/src/test/java/org/apache/hadoop/hbase/avro/TestAvroServer.java
index 5232aa4055c..23b91e12d0f 100644
--- a/src/test/java/org/apache/hadoop/hbase/avro/TestAvroServer.java
+++ b/src/test/java/org/apache/hadoop/hbase/avro/TestAvroServer.java
@@ -171,7 +171,6 @@ public class TestAvroServer {
     impl.deleteFamily(tableAname, familyAname);
     assertEquals(impl.describeTable(tableAname).families.size(), 0);
 
-    impl.disableTable(tableAname);
     impl.deleteTable(tableAname);
   }
 
diff --git a/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java b/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java
index 9ef8ad48654..c6c0d2baabf 100644
--- a/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java
+++ b/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java
@@ -1,5 +1,5 @@
 /**
- * Copyright 2009 The Apache Software Foundation
+ * Copyright 2011 The Apache Software Foundation
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -46,6 +46,7 @@ import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.NotServingRegionException;
 import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableNotDisabledException;
+import org.apache.hadoop.hbase.TableNotEnabledException;
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.executor.EventHandler;
 import org.apache.hadoop.hbase.executor.EventHandler.EventType;
@@ -916,12 +917,37 @@ public class TestAdmin {
    * @throws IOException
    */
   @Test (expected=TableExistsException.class)
-  public void testTableNotFoundExceptionWithATable() throws IOException {
-    final byte [] name = Bytes.toBytes("testTableNotFoundExceptionWithATable");
+  public void testTableExistsExceptionWithATable() throws IOException {
+    final byte [] name = Bytes.toBytes("testTableExistsExceptionWithATable");
     TEST_UTIL.createTable(name, HConstants.CATALOG_FAMILY);
     TEST_UTIL.createTable(name, HConstants.CATALOG_FAMILY);
   }
 
+  /**
+   * Can't disable a table if the table isn't in enabled state
+   * @throws IOException
+   */
+  @Test (expected=TableNotEnabledException.class)
+  public void testTableNotEnabledExceptionWithATable() throws IOException {
+    final byte [] name = Bytes.toBytes(
+      "testTableNotEnabledExceptionWithATable");
+    TEST_UTIL.createTable(name, HConstants.CATALOG_FAMILY);
+    this.admin.disableTable(name);
+    this.admin.disableTable(name);
+  }
+
+  /**
+   * Can't enable a table if the table isn't in disabled state
+   * @throws IOException
+   */
+  @Test (expected=TableNotDisabledException.class)
+  public void testTableNotDisabledExceptionWithATable() throws IOException {
+    final byte [] name = Bytes.toBytes(
+      "testTableNotDisabledExceptionWithATable");
+    TEST_UTIL.createTable(name, HConstants.CATALOG_FAMILY);
+    this.admin.enableTable(name);
+  }
+
   /**
    * For HADOOP-2579
    * @throws IOException
diff --git a/src/test/java/org/apache/hadoop/hbase/rest/TestSchemaResource.java b/src/test/java/org/apache/hadoop/hbase/rest/TestSchemaResource.java
index 96791e5f59c..14d24bab53f 100644
--- a/src/test/java/org/apache/hadoop/hbase/rest/TestSchemaResource.java
+++ b/src/test/java/org/apache/hadoop/hbase/rest/TestSchemaResource.java
@@ -104,9 +104,6 @@ public class TestSchemaResource {
     response = client.put(schemaPath, Constants.MIMETYPE_XML, toXML(model));
     assertEquals(response.getCode(), 403);
 
-    // make sure HBase concurs, and wait for the table to come online
-    admin.enableTable(TABLE1);
-
     // retrieve the schema and validate it
     response = client.get(schemaPath, Constants.MIMETYPE_XML);
     assertEquals(response.getCode(), 200);
@@ -145,9 +142,6 @@ public class TestSchemaResource {
       model.createProtobufOutput());
     assertEquals(response.getCode(), 403);
 
-    // make sure HBase concurs, and wait for the table to come online
-    admin.enableTable(TABLE2);
-
     // retrieve the schema and validate it
     response = client.get(schemaPath, Constants.MIMETYPE_PROTOBUF);
     assertEquals(response.getCode(), 200);