Second batch.

Author: viirya

druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java

@@ -294,6 +294,11 @@ public String call() throws Exception {
             );
           }
         }, new Predicate<Throwable>() {
+          @Override
+          public boolean test(@Nullable Throwable input) {
+            return input instanceof IOException;
+          }
+
           @Override
           public boolean apply(@Nullable Throwable input) {
             return input instanceof IOException;
@@ -341,6 +346,18 @@ public URL apply(DataSegment dataSegment) {
       int numRetries = 0;
       while (numRetries++ < maxTries && !setOfUrls.isEmpty()) {
         setOfUrls = ImmutableSet.copyOf(Sets.filter(setOfUrls, new Predicate<URL>() {
+          @Override
+          public boolean test(URL input) {
+            try {
+              String result = DruidStorageHandlerUtils.getURL(httpClient, input);
+              LOG.debug(String.format("Checking segment [%s] response is [%s]", input, result));
+              return Strings.isNullOrEmpty(result);
+            } catch (IOException e) {
+              LOG.error(String.format("Error while checking URL [%s]", input), e);
+              return true;
+            }
+          }
+
           @Override
           public boolean apply(URL input) {
             try {

druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java

@@ -17,7 +17,6 @@
  */
 package org.apache.hadoop.hive.druid.serde;
 
-import com.google.common.collect.Iterators;
 import com.metamx.common.lifecycle.Lifecycle;
 import com.metamx.http.client.HttpClient;
 import com.metamx.http.client.HttpClientConfig;
@@ -37,6 +36,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -63,7 +63,8 @@ public abstract class DruidQueryRecordReader<T extends BaseQuery<R>, R extends C
   /**
    * Query results.
    */
-  protected Iterator<R> results = Iterators.emptyIterator();
+
+  protected Iterator<R> results = Collections.emptyIterator();
 
   @Override
   public void initialize(InputSplit split, TaskAttemptContext context) throws IOException {

druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSelectQueryRecordReader.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -27,7 +28,6 @@
 import org.apache.hadoop.io.NullWritable;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.google.common.collect.Iterators;
 
 import io.druid.query.Result;
 import io.druid.query.select.EventHolder;
@@ -42,7 +42,7 @@ public class DruidSelectQueryRecordReader
 
   private Result<SelectResultValue> current;
 
-  private Iterator<EventHolder> values = Iterators.emptyIterator();
+  private Iterator<EventHolder> values = Collections.emptyIterator();
 
   @Override
   protected SelectQuery createQuery(String content) throws IOException {

druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidTopNQueryRecordReader.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -27,7 +28,6 @@
 import org.apache.hadoop.io.NullWritable;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.google.common.collect.Iterators;
 
 import io.druid.query.Result;
 import io.druid.query.topn.DimensionAndMetricValueExtractor;
@@ -42,7 +42,7 @@ public class DruidTopNQueryRecordReader
 
   private Result<TopNResultValue> current;
 
-  private Iterator<DimensionAndMetricValueExtractor> values = Iterators.emptyIterator();
+  private Iterator<DimensionAndMetricValueExtractor> values = Collections.emptyIterator();
 
   @Override
   protected TopNQuery createQuery(String content) throws IOException {

itests/hive-unit/src/test/java/org/apache/hadoop/hive/io/TestHadoopFileStatus.java

@@ -79,6 +79,14 @@ public void testHadoopFileStatusAclEntries() throws IOException {
     Assert.assertNotNull(sourceStatus.getAclEntries());
     Assert.assertTrue(sourceStatus.getAclEntries().size() == 3);
     Iterables.removeIf(sourceStatus.getAclEntries(), new Predicate<AclEntry>() {
+      @Override
+      public boolean test(AclEntry input) {
+        if (input.getName() == null) {
+          return true;
+        }
+        return false;
+      }
+
       @Override
       public boolean apply(AclEntry input) {
         if (input.getName() == null) {

itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestServiceDiscovery.java

@@ -143,6 +143,12 @@ private class ConnParamInfoPred implements Predicate<ConnParamInfo> {
       this.pathPrefix = pathPrefix;
     }
 
+    @Override
+    public boolean test(ConnParamInfo inputParam) {
+      return inputParam.host.equals(host) && inputParam.port == port &&
+              inputParam.path.startsWith(pathPrefix);
+    }
+
     @Override
     public boolean apply(ConnParamInfo inputParam) {
       return inputParam.host.equals(host) && inputParam.port == port &&

itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreCliDriver.java

@@ -153,7 +153,7 @@ public Void invokeInternal() throws Exception {
 
   @Override
   public void runTest(String tname, String fname, String fpath) throws Exception {
-    Stopwatch sw = new Stopwatch().start();
+    Stopwatch sw = Stopwatch.createStarted();
     boolean skipped = false;
     boolean failed = false;
     try {

itests/util/src/main/java/org/apache/hadoop/hive/ql/security/authorization/plugin/sqlstd/SQLStdHiveAuthorizationValidatorForTest.java

@@ -79,6 +79,15 @@ List<HivePrivilegeObject> filterForBypass(List<HivePrivilegeObject> privilegeObj
       return null;
     } else {
       return Lists.newArrayList(Iterables.filter(privilegeObjects,new Predicate<HivePrivilegeObject>() {
+        @Override
+        public boolean test(@Nullable HivePrivilegeObject hivePrivilegeObject) {
+          // Return true to retain an item, and false to filter it out.
+          if (hivePrivilegeObject == null){
+            return true;
+          }
+          return !bypassObjectTypes.contains(hivePrivilegeObject.getType());
+        }
+
         @Override
         public boolean apply(@Nullable HivePrivilegeObject hivePrivilegeObject) {
           // Return true to retain an item, and false to filter it out.

itests/util/src/main/java/org/apache/hadoop/hive/util/ElapsedTimeLoggingWrapper.java

@@ -28,7 +28,7 @@ public abstract class ElapsedTimeLoggingWrapper<T> {
   public abstract T invokeInternal() throws Exception;
 
   public T invoke(String message, Logger LOG, boolean toStdErr) throws Exception {
-    Stopwatch sw = new Stopwatch().start();
+    Stopwatch sw = Stopwatch.createStarted();
     try {
       T retVal = invokeInternal();
       return retVal;

llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java

@@ -176,7 +176,7 @@ public void onFailure(Throwable t) {
           Thread.getDefaultUncaughtExceptionHandler().uncaughtException(Thread.currentThread(), t);
         }
       }
-    });
+    }, queueLookupExecutor);
     // TODO: why is this needed? we could just save the host and port?
     nodeId = LlapNodeId.getInstance(localAddress.get().getHostName(), localAddress.get().getPort());
     LOG.info("AMReporter running with DaemonId: {}, NodeId: {}", daemonId, nodeId);
@@ -271,7 +271,7 @@ public void onFailure(Throwable t) {
         LOG.warn("Failed to send taskKilled for {}. The attempt will likely time out.",
             taskAttemptId);
       }
-    });
+    }, executor);
   }
 
   public void queryComplete(QueryIdentifier queryIdentifier) {
@@ -337,7 +337,7 @@ public void onFailure(Throwable t) {
                     amNodeInfo.amNodeId, currentQueryIdentifier, t);
                   queryFailedHandler.queryFailed(currentQueryIdentifier);
                 }
-              });
+              }, executor);
             }
           }
         } catch (InterruptedException e) {