Sequences refactorings and removed unused code (part of apache#3798) (a…

…pache#3693)

* Removing unused code from io.druid.java.util.common.guava package; fix apache#3563 (more consistent and paranoiac resource handing in Sequences subsystem); Add Sequences.wrap() for DRY in MetricsEmittingQueryRunner, CPUTimeMetricQueryRunner and SpecificSegmentQueryRunner; Catch MissingSegmentsException in SpecificSegmentQueryRunner's yielder.next() method (follow up on apache#3617)

* Make Sequences.withEffect() execute the effect if the wrapped sequence throws exception from close()

* Fix strange code in MetricsEmittingQueryRunner

* Add comment on why YieldingSequenceBase is used in Sequences.withEffect()

* Use Closer in OrderedMergeSequence and MergeSequence to close multiple yielders

common/src/main/java/io/druid/collections/OrderedMergeSequence.java

@@ -23,6 +23,7 @@
 import com.google.common.base.Throwables;
 import com.google.common.collect.Ordering;
 
+import com.google.common.io.Closer;
 import io.druid.java.util.common.guava.Accumulator;
 import io.druid.java.util.common.guava.CloseQuietly;
 import io.druid.java.util.common.guava.Sequence;
@@ -207,9 +208,11 @@ public boolean isDone()
       @Override
       public void close() throws IOException
       {
+        Closer closer = Closer.create();
         while(!pQueue.isEmpty()) {
-          pQueue.remove().close();
+          closer.register(pQueue.remove());
         }
+        closer.close();
       }
     };
   }

common/src/test/java/io/druid/common/guava/CombiningSequenceTest.java

@@ -24,15 +24,12 @@
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Ordering;
-
 import io.druid.java.util.common.Pair;
-import io.druid.java.util.common.guava.ResourceClosingSequence;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.guava.Yielder;
 import io.druid.java.util.common.guava.YieldingAccumulator;
 import io.druid.java.util.common.guava.nary.BinaryFn;
-
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -232,7 +229,7 @@ public void close() throws IOException
 
     Sequence<Pair<Integer, Integer>> seq = Sequences.limit(
         new CombiningSequence<>(
-            new ResourceClosingSequence<>(Sequences.simple(pairs), closeable),
+            Sequences.withBaggage(Sequences.simple(pairs), closeable),
             Ordering.natural().onResultOf(Pair.<Integer, Integer>lhsFn()),
             new BinaryFn<Pair<Integer, Integer>, Pair<Integer, Integer>, Pair<Integer, Integer>>()
             {

java-util/src/main/java/io/druid/java/util/common/guava/BaseSequence.java

@@ -19,9 +19,6 @@
 
 package io.druid.java.util.common.guava;
 
-import com.google.common.base.Throwables;
-import io.druid.java.util.common.logger.Logger;
-
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Iterator;
@@ -30,30 +27,9 @@
  */
 public class BaseSequence<T, IterType extends Iterator<T>> implements Sequence<T>
 {
-  private static final Logger log = new Logger(BaseSequence.class);
 
   private final IteratorMaker<T, IterType> maker;
 
-  public static <T> Sequence<T> simple(final Iterable<T> iterable)
-  {
-    return new BaseSequence<>(
-        new BaseSequence.IteratorMaker<T, Iterator<T>>()
-        {
-          @Override
-          public Iterator<T> make()
-          {
-            return iterable.iterator();
-          }
-
-          @Override
-          public void cleanup(Iterator<T> iterFromMake)
-          {
-
-          }
-        }
-    );
-  }
-
   public BaseSequence(
       IteratorMaker<T, IterType> maker
   )
@@ -69,11 +45,18 @@ public <OutType> OutType accumulate(OutType initValue, final Accumulator<OutType
       while (iterator.hasNext()) {
         initValue = fn.accumulate(initValue, iterator.next());
       }
-      return initValue;
     }
-    finally {
-      maker.cleanup(iterator);
+    catch (Throwable t) {
+      try {
+        maker.cleanup(iterator);
+      }
+      catch (Exception e) {
+        t.addSuppressed(e);
+      }
+      throw t;
     }
+    maker.cleanup(iterator);
+    return initValue;
   }
 
   @Override
@@ -84,16 +67,14 @@ public <OutType> Yielder<OutType> toYielder(OutType initValue, YieldingAccumulat
     try {
       return makeYielder(initValue, accumulator, iterator);
     }
-    catch (Exception e) {
-      // We caught an Exception instead of returning a really, real, live, real boy, errr, iterator
-      // So we better try to close our stuff, 'cause the exception is what is making it out of here.
+    catch (Throwable t) {
       try {
         maker.cleanup(iterator);
       }
-      catch (RuntimeException e1) {
-        log.error(e1, "Exception thrown when closing maker.  Logging and ignoring.");
+      catch (Exception e) {
+        t.addSuppressed(e);
       }
-      throw Throwables.propagate(e);
+      throw t;
     }
   }
 
@@ -138,16 +119,14 @@ public Yielder<OutType> next(OutType initValue)
         try {
           return makeYielder(initValue, accumulator, iter);
         }
-        catch (Exception e) {
-          // We caught an Exception instead of returning a really, real, live, real boy, errr, iterator
-          // So we better try to close our stuff, 'cause the exception is what is making it out of here.
+        catch (Throwable t) {
           try {
             maker.cleanup(iter);
           }
-          catch (RuntimeException e1) {
-            log.error(e1, "Exception thrown when closing maker.  Logging and ignoring.");
+          catch (Exception e) {
+            t.addSuppressed(e);
           }
-          throw Throwables.propagate(e);
+          throw t;
         }
       }
 
@@ -165,10 +144,10 @@ public void close() throws IOException
     };
   }
 
-  public static interface IteratorMaker<T, IterType extends Iterator<T>>
+  public interface IteratorMaker<T, IterType extends Iterator<T>>
   {
-    public IterType make();
+    IterType make();
 
-    public void cleanup(IterType iterFromMake);
+    void cleanup(IterType iterFromMake);
   }
 }

java-util/src/main/java/io/druid/java/util/common/guava/ConcatSequence.java

@@ -21,6 +21,7 @@
 
 import com.google.common.base.Throwables;
 
+import java.io.Closeable;
 import java.io.IOException;
 
 /**
@@ -73,11 +74,14 @@ public Sequence<T> accumulate(Sequence<T> accumulated, Sequence<T> in)
     try {
       return makeYielder(yielderYielder, initValue, accumulator);
     }
-    catch (RuntimeException e) {
-      // We caught a RuntimeException instead of returning a really, real, live, real boy, errr, iterator
-      // So we better try to close our stuff, 'cause the exception is what is making it out of here.
-      CloseQuietly.close(yielderYielder);
-      throw e;
+    catch (Throwable t) {
+      try {
+        yielderYielder.close();
+      }
+      catch (Exception e) {
+        t.addSuppressed(e);
+      }
+      throw t;
     }
   }
 
@@ -87,10 +91,6 @@ public <OutType> Yielder<OutType> makeYielder(
       YieldingAccumulator<OutType, T> accumulator
   )
   {
-    if (yielderYielder.isDone()) {
-      return Yielders.done(initValue, yielderYielder);
-    }
-
     while (!yielderYielder.isDone()) {
       Yielder<OutType> yielder = yielderYielder.get().toYielder(initValue, accumulator);
       if (accumulator.yielded()) {
@@ -152,8 +152,9 @@ public boolean isDone()
       @Override
       public void close() throws IOException
       {
-        yielder.close();
-        yielderYielder.close();
+        try (Closeable toClose = yielderYielder) {
+          yielder.close();
+        }
       }
     };
   }

java-util/src/main/java/io/druid/java/util/common/guava/ExecuteWhenDoneYielder.java

@@ -56,9 +56,28 @@ public boolean isDone()
   @Override
   public void close() throws IOException
   {
-    if (isDone()) {
-      executor.execute(runnable);
+    boolean done = isDone();
+    Throwable thrown = null;
+    try {
+      baseYielder.close();
+    }
+    catch (Throwable t) {
+      thrown = t;
+      throw t;
+    }
+    finally {
+      if (done) {
+        if (thrown != null) {
+          try {
+            executor.execute(runnable);
+          }
+          catch (Throwable t) {
+            thrown.addSuppressed(t);
+          }
+        } else {
+          executor.execute(runnable);
+        }
+      }
     }
-    baseYielder.close();
   }
 }