同步官方最新代码

bin/hbase

@@ -92,7 +92,8 @@ if [ $# = 0 ]; then
   echo "Commands:"
   echo "Some commands take arguments. Pass no args or -h for usage."
   echo "  shell           Run the HBase shell"
-  echo "  hbck            Run the hbase 'fsck' tool"
+  echo "  hbck            Run the HBase 'fsck' tool. Defaults read-only hbck1."
+  echo "                  Pass '-j /path/to/HBCK2.jar' to run hbase-2.x HBCK2."
   echo "  snapshot        Tool for managing snapshots"
   if [ "${in_omnibus_tarball}" = "true" ]; then
     echo "  wal             Write-ahead-log analyzer"
@@ -482,7 +483,25 @@ if [ "$COMMAND" = "shell" ] ; then
   HBASE_OPTS="$HBASE_OPTS $HBASE_SHELL_OPTS"
   CLASS="org.jruby.Main -X+O ${JRUBY_OPTS} ${HBASE_HOME}/bin/hirb.rb"
 elif [ "$COMMAND" = "hbck" ] ; then
-  CLASS='org.apache.hadoop.hbase.util.HBaseFsck'
+  # Look for the -j /path/to/HBCK2.jar parameter. Else pass through to hbck.
+  case "${1}" in
+    -j)
+    # Found -j parameter. Add arg to CLASSPATH and set CLASS to HBCK2.
+    shift
+    JAR="${1}"
+    if [ ! -f "${JAR}" ]; then
+      echo "${JAR} file not found!"
+      echo "Usage: hbase [<options>] hbck -jar /path/to/HBCK2.jar [<args>]"
+      exit 1
+    fi
+    CLASSPATH="${JAR}:${CLASSPATH}";
+    CLASS="org.apache.hbase.HBCK2"
+    shift # past argument=value
+    ;;
+    *)
+    CLASS='org.apache.hadoop.hbase.util.HBaseFsck'
+    ;;
+  esac
 elif [ "$COMMAND" = "wal" ] ; then
   CLASS='org.apache.hadoop.hbase.wal.WALPrettyPrinter'
 elif [ "$COMMAND" = "hfile" ] ; then

dev-support/Jenkinsfile

@@ -80,7 +80,8 @@ pipeline {
               if [[ true !=  "${USE_YETUS_PRERELEASE}" ]]; then
                 YETUS_DIR="${WORKSPACE}/yetus-${YETUS_RELEASE}"
                 echo "Checking for Yetus ${YETUS_RELEASE} in '${YETUS_DIR}'"
-                if [ ! -d "${YETUS_DIR}" ]; then
+                if ! "${YETUS_DIR}/bin/test-patch" --version >/dev/null 2>&1 ; then
+                  rm -rf "${YETUS_DIR}"
                   "${WORKSPACE}/component/dev-support/jenkins-scripts/cache-apache-project-artifact.sh" \
                       --working-dir "${WORKSPACE}/downloads-yetus" \
                       --keys 'https://www.apache.org/dist/yetus/KEYS' \

dev-support/submit-patch.py

@@ -171,17 +171,16 @@ def validate_patch_dir(patch_dir):
 # - current branch is same as base branch
 # - current branch is ahead of base_branch by more than 1 commits
 def check_diff_between_branches(base_branch):
-    only_in_base_branch = git.log("HEAD.." + base_branch, oneline = True)
-    only_in_active_branch = git.log(base_branch + "..HEAD", oneline = True)
+    only_in_base_branch = list(repo.iter_commits("HEAD.." + base_branch))
+    only_in_active_branch = list(repo.iter_commits(base_branch + "..HEAD"))
     if len(only_in_base_branch) != 0:
         log_fatal_and_exit(" '%s' is ahead of current branch by %s commits. Rebase "
-                           "and try again.", base_branch, len(only_in_base_branch.split("\n")))
+                           "and try again.", base_branch, len(only_in_base_branch))
     if len(only_in_active_branch) == 0:
         log_fatal_and_exit(" Current branch is same as '%s'. Exiting...", base_branch)
-    if len(only_in_active_branch.split("\n")) > 1:
+    if len(only_in_active_branch) > 1:
         log_fatal_and_exit(" Current branch is ahead of '%s' by %s commits. Squash into single "
-                           "commit and try again.",
-                           base_branch, len(only_in_active_branch.split("\n")))
+                           "commit and try again.", base_branch, len(only_in_active_branch))
 
 
 # If ~/.apache-creds is present, load credentials from it otherwise prompt user.
@@ -277,7 +276,7 @@ def get_review_board_id_if_present(issue_url, rb_link_title):
             # Use jira summary as review's summary too.
             summary = get_jira_summary(issue_url)
             # Use commit message as description.
-            description = git.log("-1", pretty="%B")
+            description = repo.head.commit.message
             update_draft_data = {"bugs_closed" : [args.jira_id.upper()], "target_groups" : "hbase",
                                  "target_people" : args.reviewers, "summary" : summary,
                                  "description" : description }

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java

@@ -296,9 +296,20 @@ private List<String> getLogFilesForNewBackup(HashMap<String, Long> olderTimestam
         currentLogFile = log.getPath().toString();
         resultLogFiles.add(currentLogFile);
         currentLogTS = BackupUtils.getCreationTime(log.getPath());
-        // newestTimestamps is up-to-date with the current list of hosts
-        // so newestTimestamps.get(host) will not be null.
-        if (currentLogTS > newestTimestamps.get(host)) {
+
+        // If newestTimestamps.get(host) is null, means that
+        // either RS (host) has been restarted recently with different port number
+        // or RS is down (was decommisioned). In any case, we treat this
+        // log file as eligible for inclusion into incremental backup log list
+        Long ts = newestTimestamps.get(host);
+        if (ts ==  null) {
+          LOG.warn("ORPHAN log found: " + log + " host=" + host);
+          LOG.debug("Known hosts (from newestTimestamps):");
+          for (String s: newestTimestamps.keySet()) {
+            LOG.debug(s);
+          }
+        }
+        if (ts == null || currentLogTS > ts) {
           newestLogs.add(currentLogFile);
         }
       }
@@ -343,7 +354,7 @@ private List<String> getLogFilesForNewBackup(HashMap<String, Long> olderTimestam
       // Even if these logs belong to a obsolete region server, we still need
       // to include they to avoid loss of edits for backup.
       Long newTimestamp = newestTimestamps.get(host);
-      if (newTimestamp != null && currentLogTS > newTimestamp) {
+      if (newTimestamp == null || currentLogTS > newTimestamp) {
         newestLogs.add(currentLogFile);
       }
     }

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/mapreduce/MapReduceHFileSplitterJob.java

@@ -164,7 +164,7 @@ public static void main(String[] args) throws Exception {
   public int run(String[] args) throws Exception {
     if (args.length < 2) {
       usage("Wrong number of arguments: " + args.length);
-      System.exit(-1);
+      return -1;
     }
     Job job = createSubmittableJob(args);
     int result = job.waitForCompletion(true) ? 0 : 1;

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java

@@ -327,7 +327,7 @@ public static void checkTargetDir(String backupRootPath, Configuration conf) thr
       if (expMsg.contains("No FileSystem for scheme")) {
         newMsg =
             "Unsupported filesystem scheme found in the backup target url. Error Message: "
-                + newMsg;
+                + expMsg;
         LOG.error(newMsg);
         throw new IOException(newMsg);
       } else {

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java

@@ -1737,11 +1737,14 @@ HTableDescriptor[] getTableDescriptors(List<String> names)
 
   /**
    * Abort a procedure.
+   * Do not use. Usually it is ignored but if not, it can do more damage than good. See hbck2.
    * @param procId ID of the procedure to abort
    * @param mayInterruptIfRunning if the proc completed at least one step, should it be aborted?
    * @return <code>true</code> if aborted, <code>false</code> if procedure already completed or does not exist
    * @throws IOException
+   * @deprecated Since 2.1.1 -- to be removed.
    */
+  @Deprecated
   boolean abortProcedure(
       long procId,
       boolean mayInterruptIfRunning) throws IOException;
@@ -1752,12 +1755,15 @@ boolean abortProcedure(
    * It may throw ExecutionException if there was an error while executing the operation
    * or TimeoutException in case the wait timeout was not long enough to allow the
    * operation to complete.
+   * Do not use. Usually it is ignored but if not, it can do more damage than good. See hbck2.
    *
    * @param procId ID of the procedure to abort
    * @param mayInterruptIfRunning if the proc completed at least one step, should it be aborted?
    * @return <code>true</code> if aborted, <code>false</code> if procedure already completed or does not exist
    * @throws IOException
+   * @deprecated Since 2.1.1 -- to be removed.
    */
+  @Deprecated
   Future<Boolean> abortProcedureAsync(
     long procId,
     boolean mayInterruptIfRunning) throws IOException;

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -877,12 +877,15 @@ CompletableFuture<Boolean> isProcedureFinished(String signature, String instance
       Map<String, String> props);
 
   /**
-   * abort a procedure
+   * Abort a procedure
+   * Do not use. Usually it is ignored but if not, it can do more damage than good. See hbck2.
    * @param procId ID of the procedure to abort
    * @param mayInterruptIfRunning if the proc completed at least one step, should it be aborted?
    * @return true if aborted, false if procedure already completed or does not exist. the value is
    *         wrapped by {@link CompletableFuture}
+   * @deprecated Since 2.1.1 -- to be removed.
    */
+  @Deprecated
   CompletableFuture<Boolean> abortProcedure(long procId, boolean mayInterruptIfRunning);
 
   /**

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRequestFutureImpl.java

@@ -826,7 +826,9 @@ private void receiveMultiAction(MultiAction multiAction,
       byte[] regionName = regionEntry.getKey();
 
       Throwable regionException = responses.getExceptions().get(regionName);
-      cleanServerCache(server, regionException);
+      if (regionException != null) {
+        cleanServerCache(server, regionException);
+      }
 
       Map<Integer, Object> regionResults =
         results.containsKey(regionName) ? results.get(regionName).result : Collections.emptyMap();

hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java

@@ -25,6 +25,7 @@
 import java.io.InterruptedIOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -230,7 +231,6 @@
  * @see Admin
  */
 @InterfaceAudience.Private
-@InterfaceStability.Evolving
 public class HBaseAdmin implements Admin {
   private static final Logger LOG = LoggerFactory.getLogger(HBaseAdmin.class);
 
@@ -4314,15 +4314,13 @@ public Void call() throws Exception {
   }
 
   @Override
-  public List<ServerName> clearDeadServers(final List<ServerName> servers) throws IOException {
-    if (servers == null || servers.size() == 0) {
-      throw new IllegalArgumentException("servers cannot be null or empty");
-    }
+  public List<ServerName> clearDeadServers(List<ServerName> servers) throws IOException {
     return executeCallable(new MasterCallable<List<ServerName>>(getConnection(),
             getRpcControllerFactory()) {
       @Override
       protected List<ServerName> rpcCall() throws Exception {
-        ClearDeadServersRequest req = RequestConverter.buildClearDeadServersRequest(servers);
+        ClearDeadServersRequest req = RequestConverter.
+          buildClearDeadServersRequest(servers == null? Collections.EMPTY_LIST: servers);
         return ProtobufUtil.toServerNameList(
                 master.clearDeadServers(getRpcController(), req).getServerNameList());
       }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseHbck.java

@@ -18,25 +18,38 @@
 package org.apache.hadoop.hbase.client;
 
 import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
-import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetTableStateResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.HbckService.BlockingInterface;
 
+import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 
 /**
  * Use {@link ClusterConnection#getHbck()} to obtain an instance of {@link Hbck} instead of
- * constructing
- * an HBaseHbck directly. This will be mostly used by hbck tool.
+ * constructing an HBaseHbck directly.
  *
  * <p>Connection should be an <i>unmanaged</i> connection obtained via
  * {@link ConnectionFactory#createConnection(Configuration)}.</p>
  *
+ * <p>NOTE: The methods in here can do damage to a cluster if applied in the wrong sequence or at
+ * the wrong time. Use with caution. For experts only. These methods are only for the
+ * extreme case where the cluster has been damaged or has achieved an inconsistent state because
+ * of some unforeseen circumstance or bug and requires manual intervention.
+ *
  * <p>An instance of this class is lightweight and not-thread safe. A new instance should be created
  * by each thread. Pooling or caching of the instance is not recommended.</p>
  *
@@ -75,10 +88,6 @@ public boolean isAborted() {
     return this.aborted;
   }
 
-  /**
-   * NOTE: This is a dangerous action, as existing running procedures for the table or regions
-   * which belong to the table may get confused.
-   */
   @Override
   public TableState setTableStateInMeta(TableState state) throws IOException {
     try {
@@ -87,9 +96,62 @@ public TableState setTableStateInMeta(TableState state) throws IOException {
           RequestConverter.buildSetTableStateInMetaRequest(state));
       return TableState.convert(state.getTableName(), response.getTableState());
     } catch (ServiceException se) {
-      LOG.debug("ServiceException while updating table state in meta. table={}, state={}",
-          state.getTableName(), state.getState());
+      LOG.debug("table={}, state={}", state.getTableName(), state.getState(), se);
+      throw new IOException(se);
+    }
+  }
+
+  @Override
+  public List<Long> assigns(List<String> encodedRegionNames, boolean override)
+      throws IOException {
+    try {
+      MasterProtos.AssignsResponse response =
+          this.hbck.assigns(rpcControllerFactory.newController(),
+              RequestConverter.toAssignRegionsRequest(encodedRegionNames, override));
+      return response.getPidList();
+    } catch (ServiceException se) {
+      LOG.debug(toCommaDelimitedString(encodedRegionNames), se);
+      throw new IOException(se);
+    }
+  }
+
+  @Override
+  public List<Long> unassigns(List<String> encodedRegionNames, boolean override)
+      throws IOException {
+    try {
+      MasterProtos.UnassignsResponse response =
+          this.hbck.unassigns(rpcControllerFactory.newController(),
+              RequestConverter.toUnassignRegionsRequest(encodedRegionNames, override));
+      return response.getPidList();
+    } catch (ServiceException se) {
+      LOG.debug(toCommaDelimitedString(encodedRegionNames), se);
       throw new IOException(se);
     }
   }
+
+  private static String toCommaDelimitedString(List<String> list) {
+    return list.stream().collect(Collectors.joining(", "));
+  }
+
+  @Override
+  public List<Boolean> bypassProcedure(List<Long> pids, long waitTime, boolean override,
+      boolean recursive)
+      throws IOException {
+    MasterProtos.BypassProcedureResponse response = ProtobufUtil.call(
+        new Callable<MasterProtos.BypassProcedureResponse>() {
+          @Override
+          public MasterProtos.BypassProcedureResponse call() throws Exception {
+            try {
+              return hbck.bypassProcedure(rpcControllerFactory.newController(),
+                  MasterProtos.BypassProcedureRequest.newBuilder().addAllProcId(pids).
+                      setWaitTime(waitTime).setOverride(override).setRecursive(recursive).build());
+            } catch (Throwable t) {
+              LOG.error(pids.stream().map(i -> i.toString()).
+                  collect(Collectors.joining(", ")), t);
+              throw t;
+            }
+          }
+        });
+    return response.getBypassedList();
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java

@@ -487,35 +487,20 @@ public static <R> void doBatchWithCallback(List<? extends Row> actions, Object[]
   }
 
   @Override
-  public void delete(final Delete delete)
-  throws IOException {
-    CancellableRegionServerCallable<SingleResponse> callable =
-        new CancellableRegionServerCallable<SingleResponse>(
-            connection, getName(), delete.getRow(), this.rpcControllerFactory.newController(),
-            writeRpcTimeoutMs, new RetryingTimeTracker().start(), delete.getPriority()) {
+  public void delete(final Delete delete) throws IOException {
+    ClientServiceCallable<Void> callable =
+        new ClientServiceCallable<Void>(this.connection, getName(), delete.getRow(),
+            this.rpcControllerFactory.newController(), delete.getPriority()) {
       @Override
-      protected SingleResponse rpcCall() throws Exception {
-        MutateRequest request = RequestConverter.buildMutateRequest(
-          getLocation().getRegionInfo().getRegionName(), delete);
-        MutateResponse response = doMutate(request);
-        return ResponseConverter.getResult(request, response, getRpcControllerCellScanner());
+      protected Void rpcCall() throws Exception {
+        MutateRequest request = RequestConverter
+            .buildMutateRequest(getLocation().getRegionInfo().getRegionName(), delete);
+        doMutate(request);
+        return null;
       }
     };
-    List<Delete> rows = Collections.singletonList(delete);
-    AsyncProcessTask task = AsyncProcessTask.newBuilder()
-            .setPool(pool)
-            .setTableName(tableName)
-            .setRowAccess(rows)
-            .setCallable(callable)
-            .setRpcTimeout(writeRpcTimeoutMs)
-            .setOperationTimeout(operationTimeoutMs)
-            .setSubmittedRows(AsyncProcessTask.SubmittedRows.ALL)
-            .build();
-    AsyncRequestFuture ars = multiAp.submit(task);
-    ars.waitUntilDone();
-    if (ars.hasError()) {
-      throw ars.getErrors();
-    }
+    rpcCallerFactory.<Void>newCaller(this.writeRpcTimeoutMs)
+        .callWithRetries(callable, this.operationTimeoutMs);
   }
 
   @Override