hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From st...@apache.org
Subject svn commit: r1482675 [3/5] - in /hbase/trunk: hbase-client/src/main/java/org/apache/hadoop/hbase/ hbase-client/src/main/java/org/apache/hadoop/hbase/client/ hbase-client/src/main/java/org/apache/hadoop/hbase/exceptions/ hbase-client/src/main/java/org/a...
Date Wed, 15 May 2013 04:24:03 GMT
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
Wed May 15 04:24:02 2013
@@ -95,6 +95,7 @@ import org.apache.hadoop.hbase.exception
 import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
 import org.apache.hadoop.hbase.exceptions.NoSuchColumnFamilyException;
 import org.apache.hadoop.hbase.exceptions.NotServingRegionException;
+import org.apache.hadoop.hbase.exceptions.RegionInRecoveryException;
 import org.apache.hadoop.hbase.exceptions.RegionTooBusyException;
 import org.apache.hadoop.hbase.exceptions.UnknownProtocolException;
 import org.apache.hadoop.hbase.exceptions.UnknownScannerException;
@@ -200,6 +201,16 @@ public class HRegion implements HeapSize
 
   protected long completeSequenceId = -1L;
 
+  /**
+   * Operation enum is used in {@link HRegion#startRegionOperation} to provide operation
context for
+   * startRegionOperation to possibly invoke different checks before any region operations.
Not all
+   * operations have to be defined here. It's only needed when a special check is need in
+   * startRegionOperation
+   */
+  protected enum Operation {
+    ANY, GET, PUT, DELETE, SCAN, APPEND, INCREMENT, SPLIT_REGION, MERGE_REGION
+  }
+
   //////////////////////////////////////////////////////////////////////////////
   // Members
   //////////////////////////////////////////////////////////////////////////////
@@ -281,6 +292,11 @@ public class HRegion implements HeapSize
   private final AtomicInteger minorInProgress = new AtomicInteger(0);
 
   /**
+   * Min sequence id stored in store files of a region when opening the region
+   */
+  private long minSeqIdForLogReplay = -1;
+
+  /**
    * @return The smallest mvcc readPoint across all the scanners in this
    * region. Writes older than this readPoint, are included  in every
    * read operation.
@@ -641,6 +657,9 @@ public class HRegion implements HeapSize
           long storeSeqIdForReplay = store.getMaxSequenceId(false);
           maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(),
               storeSeqIdForReplay);
+          if (this.minSeqIdForLogReplay == -1 || storeSeqIdForReplay < this.minSeqIdForLogReplay)
{
+            this.minSeqIdForLogReplay = storeSeqIdForReplay;
+          }
           // Include bulk loaded files when determining seqIdForAssignment
           long storeSeqIdForAssignment = store.getMaxSequenceId(true);
           if (maxSeqId == -1 || storeSeqIdForAssignment > maxSeqId) {
@@ -778,6 +797,21 @@ public class HRegion implements HeapSize
     return this.closing.get();
   }
 
+  /**
+   * Reset recovering state of current region
+   * @param newState
+   */
+  public void setRecovering(boolean newState) {
+    this.getRegionInfo().setRecovering(newState);
+  }
+  
+  /**
+   * @return True if current region is in recovering
+   */
+  public boolean isRecovering() {
+    return this.getRegionInfo().isRecovering();
+  }
+
   /** @return true if region is available (not closed and not closing) */
   public boolean isAvailable() {
     return !isClosed() && !isClosing();
@@ -1608,7 +1642,7 @@ public class HRegion implements HeapSize
     // look across all the HStores for this region and determine what the
     // closest key is across all column families, since the data may be sparse
     checkRow(row, "getClosestRowBefore");
-    startRegionOperation();
+    startRegionOperation(Operation.GET);
     this.readRequestsCount.increment();
     try {
       Store store = getStore(family);
@@ -1654,7 +1688,7 @@ public class HRegion implements HeapSize
 
   protected RegionScanner getScanner(Scan scan,
       List<KeyValueScanner> additionalScanners) throws IOException {
-    startRegionOperation();
+    startRegionOperation(Operation.SCAN);
     try {
       // Verify families are all valid
       prepareScanner(scan);
@@ -1705,7 +1739,7 @@ public class HRegion implements HeapSize
   throws IOException {
     checkReadOnly();
     checkResources();
-    startRegionOperation();
+    startRegionOperation(Operation.DELETE);
     this.writeRequestsCount.increment();
     try {
       delete.getRow();
@@ -1804,7 +1838,7 @@ public class HRegion implements HeapSize
     // read lock, resources may run out.  For now, the thought is that this
     // will be extremely rare; we'll deal with it when it happens.
     checkResources();
-    startRegionOperation();
+    startRegionOperation(Operation.PUT);
     this.writeRequestsCount.increment();
     try {
       // All edits for the given row (across all column families) must happen atomically.
@@ -1862,13 +1896,29 @@ public class HRegion implements HeapSize
    */
   public OperationStatus[] batchMutate(
       Pair<Mutation, Integer>[] mutationsAndLocks) throws IOException {
+    return batchMutate(mutationsAndLocks, false);
+  }
+ 
+  /**
+   * Perform a batch of mutations.
+   * It supports only Put and Delete mutations and will ignore other types passed.
+   * @param mutationsAndLocks
+   *          the list of mutations paired with their requested lock IDs.
+   * @return an array of OperationStatus which internally contains the
+   *         OperationStatusCode and the exceptionMessage if any.
+   * @throws IOException
+   */
+  OperationStatus[] batchMutate(Pair<Mutation, Integer>[] mutationsAndLocks, boolean
isReplay)
+      throws IOException {
     BatchOperationInProgress<Pair<Mutation, Integer>> batchOp =
       new BatchOperationInProgress<Pair<Mutation,Integer>>(mutationsAndLocks);
 
     boolean initialized = false;
 
     while (!batchOp.isDone()) {
-      checkReadOnly();
+      if (!isReplay) {
+        checkReadOnly();
+      }
       checkResources();
 
       long newSize;
@@ -1876,11 +1926,13 @@ public class HRegion implements HeapSize
 
       try {
         if (!initialized) {
-          this.writeRequestsCount.increment();
-          doPreMutationHook(batchOp);
+          if (!isReplay) {
+            this.writeRequestsCount.increment();
+            doPreMutationHook(batchOp);
+          }
           initialized = true;
         }
-        long addedSize = doMiniBatchMutation(batchOp);
+        long addedSize = doMiniBatchMutation(batchOp, isReplay);
         newSize = this.addAndGetGlobalMemstoreSize(addedSize);
       } finally {
         closeRegionOperation();
@@ -1891,6 +1943,7 @@ public class HRegion implements HeapSize
     }
     return batchOp.retCodeDetails;
   }
+  
 
   private void doPreMutationHook(BatchOperationInProgress<Pair<Mutation, Integer>>
batchOp)
       throws IOException {
@@ -1927,10 +1980,9 @@ public class HRegion implements HeapSize
     }
   }
 
-
   @SuppressWarnings("unchecked")
-  private long doMiniBatchMutation(
-    BatchOperationInProgress<Pair<Mutation, Integer>> batchOp) throws IOException
{
+  private long doMiniBatchMutation(BatchOperationInProgress<Pair<Mutation, Integer>>
batchOp,
+      boolean isInReplay) throws IOException {
 
     // variable to note if all Put items are for the same CF -- metrics related
     boolean putsCfSetConsistent = true;
@@ -1941,7 +1993,7 @@ public class HRegion implements HeapSize
     //The set of columnFamilies first seen for Delete.
     Set<byte[]> deletesCfSet = null;
 
-    WALEdit walEdit = new WALEdit();
+    WALEdit walEdit = new WALEdit(isInReplay);
     MultiVersionConsistencyControl.WriteEntry w = null;
     long txid = 0;
     boolean walSyncSuccessful = false;
@@ -1983,7 +2035,11 @@ public class HRegion implements HeapSize
         try {
           if (isPutMutation) {
             // Check the families in the put. If bad, skip this one.
-            checkFamilies(familyMap.keySet());
+            if (isInReplay) {
+              removeNonExistentColumnFamilyForReplay(familyMap);
+            } else {
+              checkFamilies(familyMap.keySet());
+            }
             checkTimestamps(mutation.getFamilyMap(), now);
           } else {
             prepareDelete((Delete) mutation);
@@ -2042,7 +2098,7 @@ public class HRegion implements HeapSize
           }
         }
       }
-
+      
       // we should record the timestamp only after we have acquired the rowLock,
       // otherwise, newer puts/deletes are not guaranteed to have a newer timestamp
       now = EnvironmentEdgeManager.currentTimeMillis();
@@ -2081,9 +2137,9 @@ public class HRegion implements HeapSize
       w = mvcc.beginMemstoreInsert();
 
       // calling the pre CP hook for batch mutation
-      if (coprocessorHost != null) {
-        MiniBatchOperationInProgress<Pair<Mutation, Integer>> miniBatchOp =
-          new MiniBatchOperationInProgress<Pair<Mutation, Integer>>(batchOp.operations,
+      if (!isInReplay && coprocessorHost != null) {
+        MiniBatchOperationInProgress<Pair<Mutation, Integer>> miniBatchOp = 
+          new MiniBatchOperationInProgress<Pair<Mutation, Integer>>(batchOp.operations,

           batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
         if (coprocessorHost.preBatchMutate(miniBatchOp)) return 0L;
       }
@@ -2168,9 +2224,9 @@ public class HRegion implements HeapSize
       }
       walSyncSuccessful = true;
       // calling the post CP hook for batch mutation
-      if (coprocessorHost != null) {
-        MiniBatchOperationInProgress<Pair<Mutation, Integer>> miniBatchOp =
-          new MiniBatchOperationInProgress<Pair<Mutation, Integer>>(batchOp.operations,
+      if (!isInReplay && coprocessorHost != null) {
+        MiniBatchOperationInProgress<Pair<Mutation, Integer>> miniBatchOp = 
+          new MiniBatchOperationInProgress<Pair<Mutation, Integer>>(batchOp.operations,

           batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
         coprocessorHost.postBatchMutate(miniBatchOp);
       }
@@ -2187,7 +2243,7 @@ public class HRegion implements HeapSize
       // STEP 9. Run coprocessor post hooks. This should be done after the wal is
       // synced so that the coprocessor contract is adhered to.
       // ------------------------------------
-      if (coprocessorHost != null) {
+      if (!isInReplay && coprocessorHost != null) {
         for (int i = firstIndex; i < lastIndexExclusive; i++) {
           // only for successful puts
           if (batchOp.retCodeDetails[i].getOperationStatusCode()
@@ -2626,6 +2682,30 @@ public class HRegion implements HeapSize
     }
   }
 
+  /**
+   * During replay, there could exist column families which are removed between region server
+   * failure and replay
+   */
+  private void removeNonExistentColumnFamilyForReplay(
+      final Map<byte[], List<? extends Cell>> familyMap) {
+    List<byte[]> nonExistentList = null;
+    for (byte[] family : familyMap.keySet()) {
+      if (!this.htableDescriptor.hasFamily(family)) {
+        if (nonExistentList == null) {
+          nonExistentList = new ArrayList<byte[]>();
+        }
+        nonExistentList.add(family);
+      }
+    }
+    if (nonExistentList != null) {
+      for (byte[] family : nonExistentList) {
+        // Perhaps schema was changed between crash and replay
+        LOG.info("No family for " + Bytes.toString(family) + " omit from reply.");
+        familyMap.remove(family);
+      }
+    }
+  }
+
   void checkTimestamps(final Map<byte[], List<? extends Cell>> familyMap,
       long now) throws FailedSanityCheckException {
     if (timestampSlop == HConstants.LATEST_TIMESTAMP) {
@@ -3480,7 +3560,7 @@ public class HRegion implements HeapSize
             "after we renewed it. Could be caused by a very slow scanner " +
             "or a lengthy garbage collection");
       }
-      startRegionOperation();
+      startRegionOperation(Operation.SCAN);
       readRequestsCount.increment();
       try {
 
@@ -4651,7 +4731,7 @@ public class HRegion implements HeapSize
 
     checkReadOnly();
     // Lock row
-    startRegionOperation();
+    startRegionOperation(Operation.APPEND);
     this.writeRequestsCount.increment();
     WriteEntry w = null;
     try {
@@ -4819,7 +4899,7 @@ public class HRegion implements HeapSize
 
     checkReadOnly();
     // Lock row
-    startRegionOperation();
+    startRegionOperation(Operation.INCREMENT);
     this.writeRequestsCount.increment();
     WriteEntry w = null;
     try {
@@ -4956,7 +5036,7 @@ public class HRegion implements HeapSize
       ClassSize.OBJECT +
       ClassSize.ARRAY +
       38 * ClassSize.REFERENCE + 2 * Bytes.SIZEOF_INT +
-      (11 * Bytes.SIZEOF_LONG) +
+      (12 * Bytes.SIZEOF_LONG) +
       Bytes.SIZEOF_BOOLEAN);
 
   public static final long DEEP_OVERHEAD = FIXED_OVERHEAD +
@@ -5236,6 +5316,37 @@ public class HRegion implements HeapSize
    */
   public void startRegionOperation()
       throws NotServingRegionException, RegionTooBusyException, InterruptedIOException {
+    startRegionOperation(Operation.ANY);
+  }
+
+  /**
+   * @param op The operation is about to be taken on the region
+   * @throws NotServingRegionException
+   * @throws RegionTooBusyException
+   * @throws InterruptedIOException
+   */
+  protected void startRegionOperation(Operation op) throws NotServingRegionException,
+      RegionTooBusyException, InterruptedIOException {
+    switch (op) {
+    case INCREMENT:
+    case APPEND:
+    case GET:
+    case SCAN:
+    case SPLIT_REGION:
+    case MERGE_REGION:
+      // when a region is in recovering state, no read, split or merge is allowed
+      if (this.isRecovering()) {
+        throw new RegionInRecoveryException(this.getRegionNameAsString()
+            + " is recovering");
+      }
+      break;
+      default:
+        break;
+    }
+    if (op == Operation.MERGE_REGION || op == Operation.SPLIT_REGION) {
+      // split or merge region doesn't need to check the closing/closed state or lock the
region
+      return;
+    }
     if (this.closing.get()) {
       throw new NotServingRegionException(getRegionNameAsString() + " is closing");
     }
@@ -5450,6 +5561,14 @@ public class HRegion implements HeapSize
   }
 
   /**
+   * Gets the min sequence number that was read from storage when this region was opened.
WAL Edits
+   * with smaller sequence number will be skipped from replay.
+   */
+  public long getMinSeqIdForLogReplay() {
+    return this.minSeqIdForLogReplay;
+  }
+
+  /**
    * @return if a given region is in compaction now.
    */
   public CompactionState getCompactionState() {

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
Wed May 15 04:24:02 2013
@@ -92,6 +92,7 @@ import org.apache.hadoop.hbase.exception
 import org.apache.hadoop.hbase.exceptions.NotServingRegionException;
 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
 import org.apache.hadoop.hbase.exceptions.RegionAlreadyInTransitionException;
+import org.apache.hadoop.hbase.exceptions.RegionInRecoveryException;
 import org.apache.hadoop.hbase.exceptions.RegionMovedException;
 import org.apache.hadoop.hbase.exceptions.RegionOpeningException;
 import org.apache.hadoop.hbase.exceptions.RegionServerRunningException;
@@ -112,6 +113,7 @@ import org.apache.hadoop.hbase.ipc.Paylo
 import org.apache.hadoop.hbase.ipc.RpcServerInterface;
 import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface;
 import org.apache.hadoop.hbase.ipc.ServerRpcController;
+import org.apache.hadoop.hbase.master.SplitLogManager;
 import org.apache.hadoop.hbase.master.TableLockManager;
 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.protobuf.ReplicationProtbufUtil;
@@ -178,6 +180,7 @@ import org.apache.hadoop.hbase.protobuf.
 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupResponse;
 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStatusService;
 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.ReportRSFatalErrorRequest;
+import org.apache.hadoop.hbase.regionserver.HRegion.Operation;
 import org.apache.hadoop.hbase.regionserver.Leases.LeaseStillHeldException;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress;
 import org.apache.hadoop.hbase.regionserver.handler.CloseMetaHandler;
@@ -204,6 +207,7 @@ import org.apache.hadoop.hbase.util.Vers
 import org.apache.hadoop.hbase.zookeeper.ClusterStatusTracker;
 import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
 import org.apache.hadoop.hbase.zookeeper.MetaRegionTracker;
+import org.apache.hadoop.hbase.zookeeper.RecoveringRegionWatcher;
 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
 import org.apache.hadoop.hbase.zookeeper.ZooKeeperNodeTracker;
@@ -214,6 +218,7 @@ import org.apache.hadoop.net.DNS;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.Stat;
 import org.cliffc.high_scale_lib.Counter;
 
 import com.google.protobuf.BlockingRpcChannel;
@@ -258,6 +263,10 @@ public class HRegionServer implements Cl
   // catalog tracker
   protected CatalogTracker catalogTracker;
 
+  // Watch if a region is out of recovering state from ZooKeeper
+  @SuppressWarnings("unused")
+  private RecoveringRegionWatcher recoveringRegionWatcher;
+
   /**
    * Go here to get table descriptors.
    */
@@ -291,6 +300,13 @@ public class HRegionServer implements Cl
    */
   protected final Map<String, InetSocketAddress[]> regionFavoredNodesMap =
       new ConcurrentHashMap<String, InetSocketAddress[]>();
+   
+  /**
+   * Set of regions currently being in recovering state which means it can accept writes(edits
from
+   * previous failed region server) but not reads. A recovering region is also an online
region.
+   */
+  protected final Map<String, HRegion> recoveringRegions = Collections
+      .synchronizedMap(new HashMap<String, HRegion>());
 
   // Leases
   protected Leases leases;
@@ -456,6 +472,9 @@ public class HRegionServer implements Cl
 
   /** Handle all the snapshot requests to this server */
   RegionServerSnapshotManager snapshotManager;
+  
+  // configuration setting on if replay WAL edits directly to another RS
+  private final boolean distributedLogReplay;
 
   // Table level lock manager for locking for region operations
   private TableLockManager tableLockManager;
@@ -547,6 +566,9 @@ public class HRegionServer implements Cl
       }
     };
     this.rsHost = new RegionServerCoprocessorHost(this, this.conf);
+
+    this.distributedLogReplay = this.conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY,

+      HConstants.DEFAULT_DISTRIBUTED_LOG_REPLAY_CONFIG);
   }
 
   /**
@@ -671,6 +693,11 @@ public class HRegionServer implements Cl
     }
     this.tableLockManager = TableLockManager.createTableLockManager(conf, zooKeeper,
         new ServerName(isa.getHostName(), isa.getPort(), startcode));
+
+    // register watcher for recovering regions
+    if(this.distributedLogReplay) {
+      this.recoveringRegionWatcher = new RecoveringRegionWatcher(this.zooKeeper, this);
+    }
   }
 
   /**
@@ -1515,8 +1542,7 @@ public class HRegionServer implements Cl
     this.rpcServer.start();
 
     // Create the log splitting worker and start it
-    this.splitLogWorker = new SplitLogWorker(this.zooKeeper,
-        this.getConfiguration(), this.getServerName(), this);
+    this.splitLogWorker = new SplitLogWorker(this.zooKeeper, this.getConfiguration(), this,
this);
     splitLogWorker.start();
   }
 
@@ -1641,6 +1667,10 @@ public class HRegionServer implements Cl
       LOG.error("No sequence number found when opening " + r.getRegionNameAsString());
       openSeqNum = 0;
     }
+
+    // Update flushed sequence id of a recovering region in ZK
+    updateRecoveringRegionLastFlushedSequenceId(r);
+
     // Update ZK, or META
     if (r.getRegionInfo().isMetaRegion()) {
       MetaRegionTracker.setMetaLocation(getZooKeeper(),
@@ -1884,14 +1914,13 @@ public class HRegionServer implements Cl
   public long getLastSequenceId(byte[] region) {
     Long lastFlushedSequenceId = -1l;
     try {
-      GetLastFlushedSequenceIdRequest req =
-        RequestConverter.buildGetLastFlushedSequenceIdRequest(region);
+      GetLastFlushedSequenceIdRequest req = RequestConverter
+          .buildGetLastFlushedSequenceIdRequest(region);
       lastFlushedSequenceId = rssStub.getLastFlushedSequenceId(null, req)
-      .getLastFlushedSequenceId();
+          .getLastFlushedSequenceId();
     } catch (ServiceException e) {
       lastFlushedSequenceId = -1l;
-      LOG.warn("Unable to connect to the master to check " +
-          "the last flushed sequence id", e);
+      LOG.warn("Unable to connect to the master to check " + "the last flushed sequence id",
e);
     }
     return lastFlushedSequenceId;
   }
@@ -1965,6 +1994,10 @@ public class HRegionServer implements Cl
     return this.stopping;
   }
 
+  public Map<String, HRegion> getRecoveringRegions() {
+    return this.recoveringRegions;
+  }
+
   /**
    *
    * @return the configuration
@@ -2651,10 +2684,12 @@ public class HRegionServer implements Cl
     try {
       requestCount.increment();
       HRegion region = getRegion(request.getRegion());
+
       GetResponse.Builder builder = GetResponse.newBuilder();
       ClientProtos.Get get = request.getGet();
       Boolean existence = null;
       Result r = null;
+
       if (request.getClosestRowBefore()) {
         if (get.getColumnCount() != 1) {
           throw new DoNotRetryIOException(
@@ -3006,7 +3041,7 @@ public class HRegionServer implements Cl
               }
               List<KeyValue> values = new ArrayList<KeyValue>();
               MultiVersionConsistencyControl.setThreadReadPoint(scanner.getMvccReadPoint());
-              region.startRegionOperation();
+              region.startRegionOperation(Operation.SCAN);
               try {
                 int i = 0;
                 synchronized(scanner) {
@@ -3450,6 +3485,10 @@ public class HRegionServer implements Cl
         removeFromMovedRegions(region.getEncodedName());
 
         if (previous == null) {
+          // check if the region to be opened is marked in recovering state in ZK
+          if (isRegionMarkedRecoveringInZK(region.getEncodedName())) {
+            this.recoveringRegions.put(region.getEncodedName(), null);
+          }
           // If there is no action in progress, we can submit a specific handler.
           // Need to pass the expected version in the constructor.
           if (region.isMetaRegion()) {
@@ -3465,6 +3504,9 @@ public class HRegionServer implements Cl
 
         builder.addOpeningState(RegionOpeningState.OPENED);
 
+      } catch (KeeperException zooKeeperEx) {
+        LOG.error("Can't retrieve recovering state from zookeeper", zooKeeperEx);
+        throw new ServiceException(zooKeeperEx);
       } catch (IOException ie) {
         LOG.warn("Failed opening region " + region.getRegionNameAsString(), ie);
         if (isBulkAssign) {
@@ -3589,6 +3631,7 @@ public class HRegionServer implements Cl
       checkOpen();
       requestCount.increment();
       HRegion region = getRegion(request.getRegion());
+      region.startRegionOperation(Operation.SPLIT_REGION);
       LOG.info("Splitting " + region.getRegionNameAsString());
       region.flushcache();
       byte[] splitPoint = null;
@@ -3621,6 +3664,8 @@ public class HRegionServer implements Cl
       HRegion regionA = getRegion(request.getRegionA());
       HRegion regionB = getRegion(request.getRegionB());
       boolean forcible = request.getForcible();
+      regionA.startRegionOperation(Operation.MERGE_REGION);
+      regionB.startRegionOperation(Operation.MERGE_REGION);
       LOG.info("Receiving merging request for  " + regionA + ", " + regionB
           + ",forcible=" + forcible);
       regionA.flushcache();
@@ -3714,8 +3759,57 @@ public class HRegionServer implements Cl
   }
 
   /**
+   * Replay the given changes when distributedLogReplay WAL edits from a failed RS. The guarantee
is
+   * that the given mutations will be durable on the receiving RS if this method returns
without any
+   * exception.
+   * @param rpcc the RPC controller
+   * @param request the request
+   * @throws ServiceException
+   */
+  @Override
+  @QosPriority(priority = HConstants.REPLAY_QOS)
+  public MultiResponse replay(final RpcController rpcc, final MultiRequest request)
+      throws ServiceException {
+    long before = EnvironmentEdgeManager.currentTimeMillis();  
+    PayloadCarryingRpcController controller = (PayloadCarryingRpcController) rpcc;
+    CellScanner cellScanner = controller != null ? controller.cellScanner() : null;
+    // Clear scanner so we are not holding on to reference across call.
+    controller.setCellScanner(null);
+    try {
+      checkOpen();
+      HRegion region = getRegion(request.getRegion());
+      MultiResponse.Builder builder = MultiResponse.newBuilder();
+      List<MutationProto> mutates = new ArrayList<MutationProto>();
+      for (ClientProtos.MultiAction actionUnion : request.getActionList()) {
+        if (actionUnion.hasMutation()) {
+          MutationProto mutate = actionUnion.getMutation();
+          MutationType type = mutate.getMutateType();
+          switch (type) {
+          case PUT:
+          case DELETE:
+            mutates.add(mutate);
+            break;
+          default:
+            throw new DoNotRetryIOException("Unsupported mutate type: " + type.name());
+          }
+        } else {
+          LOG.warn("Error: invalid action: " + actionUnion + ". " + "it must be a Mutation.");
+          throw new DoNotRetryIOException("Invalid action, " + "it must be a Mutation.");
+        }
+      }
+      if (!mutates.isEmpty()) {
+        doBatchOp(builder, region, mutates, cellScanner, true);
+      }
+      return builder.build();
+    } catch (IOException ie) {
+      throw new ServiceException(ie);
+    } finally {
+      metricsRegionServer.updateReplay(EnvironmentEdgeManager.currentTimeMillis() - before);

+    }
+  }
+
+  /**
    * Roll the WAL writer of the region server.
-   *
    * @param controller the RPC controller
    * @param request the request
    * @throws ServiceException
@@ -3843,13 +3937,21 @@ public class HRegionServer implements Cl
 
   /**
    * Execute a list of Put/Delete mutations.
+   */
+  protected void doBatchOp(final MultiResponse.Builder builder,
+      final HRegion region, final List<MutationProto> mutates, final CellScanner cells)
{
+    doBatchOp(builder, region, mutates, cells, false);
+  }
+  
+  /**
+   * Execute a list of Put/Delete mutations.
    *
    * @param builder
    * @param region
    * @param mutations
    */
   protected void doBatchOp(final MultiResponse.Builder builder, final HRegion region,
-      final List<MutationProto> mutations, final CellScanner cells) {
+      final List<MutationProto> mutations, final CellScanner cells, boolean isReplay)
{
     @SuppressWarnings("unchecked")
     Pair<Mutation, Integer>[] mutationsWithLocks = new Pair[mutations.size()];
     long before = EnvironmentEdgeManager.currentTimeMillis();
@@ -3877,7 +3979,7 @@ public class HRegionServer implements Cl
         cacheFlusher.reclaimMemStoreMemory();
       }
 
-      OperationStatus codes[] = region.batchMutate(mutationsWithLocks);
+      OperationStatus codes[] = region.batchMutate(mutationsWithLocks, isReplay);
       for (i = 0; i < codes.length; i++) {
         switch (codes[i].getOperationStatusCode()) {
           case BAD_FAMILY:
@@ -4097,4 +4199,91 @@ public class HRegionServer implements Cl
   public CompactSplitThread getCompactSplitThread() {
     return this.compactSplitThread;
   }
+
+  /**
+   * check if /hbase/recovering-regions/<current region encoded name> exists. Returns
true if exists
+   * and set watcher as well.
+   * @param regionEncodedName region encode name
+   * @return true when /hbase/recovering-regions/<current region encoded name> exists
+   * @throws KeeperException
+   */
+  private boolean isRegionMarkedRecoveringInZK(String regionEncodedName) throws KeeperException
{
+    boolean result = false;
+    String nodePath = ZKUtil.joinZNode(this.zooKeeper.recoveringRegionsZNode, regionEncodedName);
+
+    byte[] node = ZKUtil.getDataAndWatch(this.zooKeeper, nodePath);
+    if (node != null) {
+      result = true;
+    }
+
+    return result;
+  }
+
+  /**
+   * A helper function to store the last flushed sequence Id with the previous failed RS
for a
+   * recovering region. The Id is used to skip wal edits which are flushed. Since the flushed
+   * sequence id is only valid for each RS, we associate the Id with corresponding failed
RS.
+   * @throws KeeperException
+   * @throws IOException
+   */
+  private void updateRecoveringRegionLastFlushedSequenceId(HRegion r) throws KeeperException,
+      IOException {
+    if (!r.isRecovering()) {
+      // return immdiately for non-recovering regions
+      return;
+    }
+
+    HRegionInfo region = r.getRegionInfo();
+    ZooKeeperWatcher zkw = getZooKeeper();
+    String previousRSName = this.getLastFailedRSFromZK(region.getEncodedName());
+    long minSeqIdForLogReplay = r.getMinSeqIdForLogReplay();
+    long lastRecordedFlushedSequenceId = -1;
+    String nodePath = ZKUtil.joinZNode(this.zooKeeper.recoveringRegionsZNode,
+      region.getEncodedName());
+    // recovering-region level
+    byte[] data = ZKUtil.getData(zkw, nodePath);
+    if (data != null) {
+      lastRecordedFlushedSequenceId = SplitLogManager.parseLastFlushedSequenceIdFrom(data);
+    }
+    if (data == null || lastRecordedFlushedSequenceId < minSeqIdForLogReplay) {
+      ZKUtil.setData(zkw, nodePath, ZKUtil.positionToByteArray(minSeqIdForLogReplay));
+    }
+    if (previousRSName != null) {
+      // one level deeper for failed RS
+      nodePath = ZKUtil.joinZNode(nodePath, previousRSName);
+      ZKUtil.setData(zkw, nodePath, ZKUtil.positionToByteArray(minSeqIdForLogReplay));
+      LOG.debug("Update last flushed sequence id of region " + region.getEncodedName() +
" for " 
+          + previousRSName);
+    } else {
+      LOG.warn("Can't find failed region server for recovering region " + region.getEncodedName());
+    }
+  }
+
+  /**
+   * Return the last failed RS name under /hbase/recovering-regions/encodedRegionName
+   * @param encodedRegionName
+   * @return
+   * @throws IOException
+   * @throws KeeperException
+   */
+  private String getLastFailedRSFromZK(String encodedRegionName) throws KeeperException {
+    String result = null;
+    long maxZxid = 0;
+    ZooKeeperWatcher zkw = this.getZooKeeper();
+    String nodePath = ZKUtil.joinZNode(zkw.recoveringRegionsZNode, encodedRegionName);
+    List<String> failedServers = ZKUtil.listChildrenNoWatch(zkw, nodePath);
+    if (failedServers == null || failedServers.isEmpty()) {
+      return result;
+    }
+    for (String failedServer : failedServers) {
+      String rsPath = ZKUtil.joinZNode(nodePath, failedServer);
+      Stat stat = new Stat();
+      ZKUtil.getDataNoWatch(zkw, rsPath, stat);
+      if (maxZxid < stat.getCzxid()) {
+        maxZxid = stat.getCzxid();
+        result = failedServer;
+      }
+    }
+    return result;
+  }
 }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LastSequenceId.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LastSequenceId.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LastSequenceId.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LastSequenceId.java
Wed May 15 04:24:02 2013
@@ -26,8 +26,8 @@ import org.apache.hadoop.classification.
 @InterfaceAudience.Private
 public interface LastSequenceId {
   /**
-   * @param regionname
-   * @return Last flushed sequence Id for regionname
+   * @param regionName Encoded region name
+   * @return Last flushed sequence Id for regionName or -1 if it can't be determined
    */
-  public long getLastSequenceId(byte[] regionname);
+  public long getLastSequenceId(byte[] regionName);
 }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java
Wed May 15 04:24:02 2013
@@ -87,4 +87,8 @@ public class MetricsRegionServer {
     }
     serverSource.updateAppend(t);
   }
+
+  public void updateReplay(long t){
+    serverSource.updateReplay(t);
+  }
 }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
Wed May 15 04:24:02 2013
@@ -19,6 +19,7 @@
 package org.apache.hadoop.hbase.regionserver;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -108,4 +109,9 @@ public interface RegionServerServices ex
    * @return The RegionServer's CatalogTracker
    */
   public CatalogTracker getCatalogTracker();
+
+  /**
+   * @return set of recovering regions on the hosting region server
+   */
+  public Map<String, HRegion> getRecoveringRegions();
 }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java
Wed May 15 04:24:02 2013
@@ -20,7 +20,11 @@ package org.apache.hadoop.hbase.regionse
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.logging.Log;
@@ -29,10 +33,12 @@ import org.apache.hadoop.classification.
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.SplitLogCounters;
 import org.apache.hadoop.hbase.SplitLogTask;
+import org.apache.hadoop.hbase.client.RetriesExhaustedException;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.exceptions.NotServingRegionException;
 import org.apache.hadoop.hbase.master.SplitLogManager;
 import org.apache.hadoop.hbase.regionserver.wal.HLogSplitter;
 import org.apache.hadoop.hbase.util.CancelableProgressable;
@@ -70,6 +76,7 @@ import org.apache.zookeeper.data.Stat;
 @InterfaceAudience.Private
 public class SplitLogWorker extends ZooKeeperListener implements Runnable {
   private static final Log LOG = LogFactory.getLog(SplitLogWorker.class);
+  private static final int checkInterval = 5000; // 5 seconds
 
   Thread worker;
   private final ServerName serverName;
@@ -83,20 +90,30 @@ public class SplitLogWorker extends ZooK
   private final Object grabTaskLock = new Object();
   private boolean workerInGrabTask = false;
   private final int report_period;
+  private RegionServerServices server = null;
 
   public SplitLogWorker(ZooKeeperWatcher watcher, Configuration conf,
-      ServerName serverName, TaskExecutor splitTaskExecutor) {
+      RegionServerServices server, TaskExecutor splitTaskExecutor) {
+    super(watcher);
+    this.server = server;
+    this.serverName = server.getServerName();
+    this.splitTaskExecutor = splitTaskExecutor;
+    report_period = conf.getInt("hbase.splitlog.report.period",
+      conf.getInt("hbase.splitlog.manager.timeout", SplitLogManager.DEFAULT_TIMEOUT) / 3);
+  }
+
+  public SplitLogWorker(ZooKeeperWatcher watcher, Configuration conf, ServerName serverName,
+      TaskExecutor splitTaskExecutor) {
     super(watcher);
     this.serverName = serverName;
     this.splitTaskExecutor = splitTaskExecutor;
     report_period = conf.getInt("hbase.splitlog.report.period",
-      conf.getInt("hbase.splitlog.manager.timeout",
-        SplitLogManager.DEFAULT_TIMEOUT) / 2);
+      conf.getInt("hbase.splitlog.manager.timeout", SplitLogManager.DEFAULT_TIMEOUT) / 3);
   }
 
-  public SplitLogWorker(ZooKeeperWatcher watcher, final Configuration conf,
-      final ServerName serverName, final LastSequenceId sequenceIdChecker) {
-    this(watcher, conf, serverName, new TaskExecutor () {
+  public SplitLogWorker(final ZooKeeperWatcher watcher, final Configuration conf,
+      RegionServerServices server, final LastSequenceId sequenceIdChecker) {
+    this(watcher, conf, server, new TaskExecutor() {
       @Override
       public Status exec(String filename, CancelableProgressable p) {
         Path rootdir;
@@ -113,7 +130,7 @@ public class SplitLogWorker extends ZooK
         // encountered a bad non-retry-able persistent error.
         try {
           if (!HLogSplitter.splitLogFile(rootdir, fs.getFileStatus(new Path(rootdir, filename)),
-            fs, conf, p, sequenceIdChecker)) {
+            fs, conf, p, sequenceIdChecker, watcher)) {
             return Status.PREEMPTED;
           }
         } catch (InterruptedIOException iioe) {
@@ -121,9 +138,18 @@ public class SplitLogWorker extends ZooK
           return Status.RESIGNED;
         } catch (IOException e) {
           Throwable cause = e.getCause();
-          if (cause instanceof InterruptedException) {
+          if (e instanceof RetriesExhaustedException && (cause instanceof NotServingRegionException

+                  || cause instanceof ConnectException 
+                  || cause instanceof SocketTimeoutException)) {
+            LOG.warn("log replaying of " + filename + " can't connect to the target regionserver,
"
+            		+ "resigning", e);
+            return Status.RESIGNED;
+          } else if (cause instanceof InterruptedException) {
             LOG.warn("log splitting of " + filename + " interrupted, resigning", e);
             return Status.RESIGNED;
+          } else if(cause instanceof KeeperException) {
+            LOG.warn("log splitting of " + filename + " hit ZooKeeper issue, resigning",
e);
+            return Status.RESIGNED;
           }
           LOG.warn("log splitting of " + filename + " failed, returning error", e);
           return Status.ERR;
@@ -204,7 +230,39 @@ public class SplitLogWorker extends ZooK
       synchronized (taskReadyLock) {
         while (seq_start == taskReadySeq) {
           try {
-            taskReadyLock.wait();
+            taskReadyLock.wait(checkInterval);
+            if (this.server != null) {
+              // check to see if we have stale recovering regions in our internal memory
state
+              Map<String, HRegion> recoveringRegions = this.server.getRecoveringRegions();
+              if (!recoveringRegions.isEmpty()) {
+                // Make a local copy to prevent ConcurrentModificationException when other
threads
+                // modify recoveringRegions
+                List<String> tmpCopy = new ArrayList<String>(recoveringRegions.keySet());
+                for (String region : tmpCopy) {
+                  String nodePath = ZKUtil.joinZNode(this.watcher.recoveringRegionsZNode,
region);
+                  try {
+                    if (ZKUtil.checkExists(this.watcher, nodePath) == -1) {
+                      HRegion r = recoveringRegions.remove(region);
+                      if (r != null) {
+                        r.setRecovering(false);
+                      }
+                      LOG.debug("Mark recovering region:" + region + " up.");
+                    } else {
+                      // current check is a defensive(or redundant) mechanism to prevent
us from
+                      // having stale recovering regions in our internal RS memory state
while
+                      // zookeeper(source of truth) says differently. We stop at the first
good one
+                      // because we should not have a single instance such as this in normal
case so
+                      // check the first one is good enough.
+                      break;
+                    }
+                  } catch (KeeperException e) {
+                    // ignore zookeeper error
+                    LOG.debug("Got a zookeeper when trying to open a recovering region",
e);
+                    break;
+                  }
+                }
+              }
+            }
           } catch (InterruptedException e) {
             LOG.info("SplitLogWorker interrupted while waiting for task," +
                 " exiting: " + e.toString() + (exitWorker ? "" :
@@ -214,6 +272,7 @@ public class SplitLogWorker extends ZooK
           }
         }
       }
+
     }
   }
 
@@ -463,9 +522,6 @@ public class SplitLogWorker extends ZooK
     }
   }
 
-
-
-
   @Override
   public void nodeDataChanged(String path) {
     // there will be a self generated dataChanged event every time attemptToOwnTask()
@@ -510,7 +566,6 @@ public class SplitLogWorker extends ZooK
     return childrenPaths;
   }
 
-
   @Override
   public void nodeChildrenChanged(String path) {
     if(path.equals(watcher.splitLogZNode)) {

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/OpenRegionHandler.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/OpenRegionHandler.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/OpenRegionHandler.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/OpenRegionHandler.java
Wed May 15 04:24:02 2013
@@ -19,6 +19,7 @@
 package org.apache.hadoop.hbase.regionserver.handler;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
@@ -137,6 +138,16 @@ public class OpenRegionHandler extends E
       if (region == null) {
         return;
       }
+
+      // check if we need set current region in recovering state
+      region.setRecovering(false);
+      Map<String, HRegion> recoveringRegions = this.rsServices.getRecoveringRegions();
+      if (recoveringRegions != null && !recoveringRegions.isEmpty()
+          && recoveringRegions.containsKey(region.getRegionInfo().getEncodedName()))
{
+        region.setRecovering(true);
+        recoveringRegions.put(region.getRegionInfo().getEncodedName(), region);
+      }
+
       boolean failed = true;
       if (tickleOpening("post_region_open")) {
         if (updateMeta(region)) {

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java?rev=1482675&r1=1482674&r2=1482675&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
Wed May 15 04:24:02 2013
@@ -1200,6 +1200,10 @@ class FSHLog implements HLog, Syncable {
       long now = EnvironmentEdgeManager.currentTimeMillis();
       // coprocessor hook:
       if (!coprocessorHost.preWALWrite(info, logKey, logEdit)) {
+        if (logEdit.isReplay()) {
+          // set replication scope null so that this won't be replicated
+          logKey.setScopes(null);
+        }
         // write to our buffer for the Hlog file.
         logSyncer.append(new FSHLog.Entry(logKey, logEdit));
       }



Mime
View raw message