Add Queues.drain(BlockingQueue, Collection, int, Duration) and drainU…

…ninterruptibly(BlockingQueue, Collection, int, Duration).

RELNOTES=add Duration-based Queues drain methods.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=247454175

guava/src/com/google/common/collect/Queues.java

@@ -252,6 +252,28 @@ public static <E> SynchronousQueue<E> newSynchronousQueue() {
     return new SynchronousQueue<E>();
   }
 
+  /**
+   * Drains the queue as {@link BlockingQueue#drainTo(Collection, int)}, but if the requested {@code
+   * numElements} elements are not available, it will wait for them up to the specified timeout.
+   *
+   * @param q the blocking queue to be drained
+   * @param buffer where to add the transferred elements
+   * @param numElements the number of elements to be waited for
+   * @param timeout how long to wait before giving up
+   * @return the number of elements transferred
+   * @throws InterruptedException if interrupted while waiting
+   * @since NEXT
+   */
+  @Beta
+  @CanIgnoreReturnValue
+  @GwtIncompatible // BlockingQueue
+  public static <E> int drain(
+      BlockingQueue<E> q, Collection<? super E> buffer, int numElements, java.time.Duration timeout)
+      throws InterruptedException {
+    // TODO(b/126049426): Consider using saturateToNanos(timeout) instead.
+    return drain(q, buffer, numElements, timeout.toNanos(), TimeUnit.NANOSECONDS);
+  }
+
   /**
    * Drains the queue as {@link BlockingQueue#drainTo(Collection, int)}, but if the requested {@code
    * numElements} elements are not available, it will wait for them up to the specified timeout.
@@ -299,6 +321,31 @@ public static <E> int drain(
     return added;
   }
 
+  /**
+   * Drains the queue as {@linkplain #drain(BlockingQueue, Collection, int, Duration)}, but with a
+   * different behavior in case it is interrupted while waiting. In that case, the operation will
+   * continue as usual, and in the end the thread's interruption status will be set (no {@code
+   * InterruptedException} is thrown).
+   *
+   * @param q the blocking queue to be drained
+   * @param buffer where to add the transferred elements
+   * @param numElements the number of elements to be waited for
+   * @param timeout how long to wait before giving up
+   * @return the number of elements transferred
+   * @since NEXT
+   */
+  @Beta
+  @CanIgnoreReturnValue
+  @GwtIncompatible // BlockingQueue
+  public static <E> int drainUninterruptibly(
+      BlockingQueue<E> q,
+      Collection<? super E> buffer,
+      int numElements,
+      java.time.Duration timeout) {
+    // TODO(b/126049426): Consider using saturateToNanos(timeout) instead.
+    return drainUninterruptibly(q, buffer, numElements, timeout.toNanos(), TimeUnit.NANOSECONDS);
+  }
+
   /**
    * Drains the queue as {@linkplain #drain(BlockingQueue, Collection, int, long, TimeUnit)}, but
    * with a different behavior in case it is interrupted while waiting. In that case, the operation