8022476: cleanup some raw types and unchecked warnings in java.util.stream

Reviewed-by: darcy
Contributed-by: mike.duigou@oracle.com, henry.jen@oracle.com

src/share/classes/java/util/Optional.java

@@ -307,7 +307,7 @@ public final class Optional<T> {
             return false;
         }
 
-        Optional other = (Optional) obj;
+        Optional<?> other = (Optional<?>) obj;
         return Objects.equals(value, other.value);
     }
 

src/share/classes/java/util/stream/AbstractPipeline.java

@@ -75,11 +75,13 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * Backlink to the head of the pipeline chain (self if this is the source
      * stage).
      */
+    @SuppressWarnings("rawtypes")
     private final AbstractPipeline sourceStage;
 
     /**
      * The "upstream" pipeline, or null if this is the source stage.
      */
+    @SuppressWarnings("rawtypes")
     private final AbstractPipeline previousStage;
 
     /**
@@ -92,6 +94,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * The next stage in the pipeline, or null if this is the last stage.
      * Effectively final at the point of linking to the next pipeline.
      */
+    @SuppressWarnings("rawtypes")
     private AbstractPipeline nextStage;
 
     /**
@@ -222,8 +225,8 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
         linkedOrConsumed = true;
 
         return isParallel()
-               ? (R) terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
-               : (R) terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
+               ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
+               : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
     }
 
     /**
@@ -232,6 +235,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * @param generator the array generator to be used to create array instances
      * @return a flat array-backed Node that holds the collected output elements
      */
+    @SuppressWarnings("unchecked")
     final Node<E_OUT> evaluateToArrayNode(IntFunction<E_OUT[]> generator) {
         if (linkedOrConsumed)
             throw new IllegalStateException("stream has already been operated upon");
@@ -256,6 +260,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * @throws IllegalStateException if this pipeline stage is not the source
      *         stage.
      */
+    @SuppressWarnings("unchecked")
     final Spliterator<E_OUT> sourceStageSpliterator() {
         if (this != sourceStage)
             throw new IllegalStateException();
@@ -265,11 +270,13 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
         linkedOrConsumed = true;
 
         if (sourceStage.sourceSpliterator != null) {
+            @SuppressWarnings("unchecked")
             Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
             sourceStage.sourceSpliterator = null;
             return s;
         }
         else if (sourceStage.sourceSupplier != null) {
+            @SuppressWarnings("unchecked")
             Spliterator<E_OUT> s = (Spliterator<E_OUT>) sourceStage.sourceSupplier.get();
             sourceStage.sourceSupplier = null;
             return s;
@@ -282,12 +289,14 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
     // BaseStream
 
     @Override
+    @SuppressWarnings("unchecked")
     public final S sequential() {
         sourceStage.parallel = false;
         return (S) this;
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     public final S parallel() {
         sourceStage.parallel = true;
         return (S) this;
@@ -295,6 +304,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
 
     // Primitive specialization use co-variant overrides, hence is not final
     @Override
+    @SuppressWarnings("unchecked")
     public Spliterator<E_OUT> spliterator() {
         if (linkedOrConsumed)
             throw new IllegalStateException("stream has already been operated upon");
@@ -302,12 +312,14 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
 
         if (this == sourceStage) {
             if (sourceStage.sourceSpliterator != null) {
-                Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
+                @SuppressWarnings("unchecked")
+                Spliterator<E_OUT> s = (Spliterator<E_OUT>) sourceStage.sourceSpliterator;
                 sourceStage.sourceSpliterator = null;
                 return s;
             }
             else if (sourceStage.sourceSupplier != null) {
-                Supplier<Spliterator<E_OUT>> s = sourceStage.sourceSupplier;
+                @SuppressWarnings("unchecked")
+                Supplier<Spliterator<E_OUT>> s = (Supplier<Spliterator<E_OUT>>) sourceStage.sourceSupplier;
                 sourceStage.sourceSupplier = null;
                 return lazySpliterator(s);
             }
@@ -349,10 +361,11 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * @param terminalFlags Operation flags for the terminal operation
      */
     private void parallelPrepare(int terminalFlags) {
+        @SuppressWarnings("rawtypes")
         AbstractPipeline backPropagationHead = sourceStage;
         if (sourceStage.sourceAnyStateful) {
             int depth = 1;
-            for (AbstractPipeline u = sourceStage, p = sourceStage.nextStage;
+            for (  @SuppressWarnings("rawtypes") AbstractPipeline u = sourceStage, p = sourceStage.nextStage;
                  p != null;
                  u = p, p = p.nextStage) {
                 int thisOpFlags = p.sourceOrOpFlags;
@@ -383,7 +396,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
         // Apply the upstream terminal flags
         if (terminalFlags != 0) {
             int upstreamTerminalFlags = terminalFlags & StreamOpFlag.UPSTREAM_TERMINAL_OP_MASK;
-            for (AbstractPipeline p = backPropagationHead; p.nextStage != null; p = p.nextStage) {
+            for ( @SuppressWarnings("rawtypes") AbstractPipeline p = backPropagationHead; p.nextStage != null; p = p.nextStage) {
                 p.combinedFlags = StreamOpFlag.combineOpFlags(upstreamTerminalFlags, p.combinedFlags);
             }
 
@@ -398,6 +411,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * of all computations up to and including the most recent stateful
      * operation.
      */
+    @SuppressWarnings("unchecked")
     private Spliterator<?> sourceSpliterator(int terminalFlags) {
         // Get the source spliterator of the pipeline
         Spliterator<?> spliterator = null;
@@ -421,7 +435,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
 
             // Adapt the source spliterator, evaluating each stateful op
             // in the pipeline up to and including this pipeline stage
-            for (AbstractPipeline u = sourceStage, p = sourceStage.nextStage, e = this;
+            for ( @SuppressWarnings("rawtypes") AbstractPipeline u = sourceStage, p = sourceStage.nextStage, e = this;
                  u != e;
                  u = p, p = p.nextStage) {
 
@@ -442,6 +456,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
 
     @Override
     final StreamShape getSourceShape() {
+        @SuppressWarnings("rawtypes")
         AbstractPipeline p = AbstractPipeline.this;
         while (p.depth > 0) {
             p = p.previousStage;
@@ -475,7 +490,9 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     final <P_IN> void copyIntoWithCancel(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
+        @SuppressWarnings({"rawtypes","unchecked"})
         AbstractPipeline p = AbstractPipeline.this;
         while (p.depth > 0) {
             p = p.previousStage;
@@ -495,16 +512,18 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
         Objects.requireNonNull(sink);
 
-        for (AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
+        for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
             sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
         }
         return (Sink<P_IN>) sink;
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     final <P_IN> Spliterator<E_OUT> wrapSpliterator(Spliterator<P_IN> sourceSpliterator) {
         if (depth == 0) {
             return (Spliterator<E_OUT>) sourceSpliterator;
@@ -591,16 +610,19 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
     /**
      * Make a node builder compatible with this stream shape.
      *
-     * @param exactSizeIfKnown if {@literal >=0}, then a node builder will be created that
-     * has a fixed capacity of at most sizeIfKnown elements. If {@literal < 0},
-     * then the node builder has an unfixed capacity. A fixed capacity node
-     * builder will throw exceptions if an element is added after builder has
-     * reached capacity, or is built before the builder has reached capacity.
+     * @param exactSizeIfKnown if {@literal >=0}, then a node builder will be
+     * created that has a fixed capacity of at most sizeIfKnown elements. If
+     * {@literal < 0}, then the node builder has an unfixed capacity. A fixed
+     * capacity node builder will throw exceptions if an element is added after
+     * builder has reached capacity, or is built before the builder has reached
+     * capacity.
+     *
      * @param generator the array generator to be used to create instances of a
      * T[] array. For implementations supporting primitive nodes, this parameter
      * may be ignored.
      * @return a node builder
      */
+    @Override
     abstract Node.Builder<E_OUT> makeNodeBuilder(long exactSizeIfKnown,
                                                  IntFunction<E_OUT[]> generator);
 
@@ -679,6 +701,7 @@ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
      * @param spliterator the source {@code Spliterator}
      * @return a {@code Spliterator} describing the result of the evaluation
      */
+    @SuppressWarnings("unchecked")
     <P_IN> Spliterator<E_OUT> opEvaluateParallelLazy(PipelineHelper<E_OUT> helper,
                                                      Spliterator<P_IN> spliterator) {
         return opEvaluateParallel(helper, spliterator, i -> (E_OUT[]) new Object[i]).spliterator();

src/share/classes/java/util/stream/AbstractShortCircuitTask.java

@@ -220,7 +220,8 @@ abstract class AbstractShortCircuitTask<P_IN, P_OUT, R,
      */
     protected void cancelLaterNodes() {
         // Go up the tree, cancel right siblings of this node and all parents
-        for (K parent = getParent(), node = (K) this; parent != null;
+        for (@SuppressWarnings("unchecked") K parent = getParent(), node = (K) this;
+             parent != null;
              node = parent, parent = parent.getParent()) {
             // If node is a left child of parent, then has a right sibling
             if (parent.leftChild == node) {

src/share/classes/java/util/stream/DoublePipeline.java

@@ -147,6 +147,7 @@ abstract class DoublePipeline<E_IN>
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     final Spliterator.OfDouble lazySpliterator(Supplier<? extends Spliterator<Double>> supplier) {
         return new StreamSpliterators.DelegatingSpliterator.OfDouble((Supplier<Spliterator.OfDouble>) supplier);
     }
@@ -209,6 +210,7 @@ abstract class DoublePipeline<E_IN>
             Sink<Double> opWrapSink(int flags, Sink<U> sink) {
                 return new Sink.ChainedDouble(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(double t) {
                         downstream.accept(mapper.apply(t));
                     }

src/share/classes/java/util/stream/IntPipeline.java

@@ -150,6 +150,7 @@ abstract class IntPipeline<E_IN>
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     final Spliterator.OfInt lazySpliterator(Supplier<? extends Spliterator<Integer>> supplier) {
         return new StreamSpliterators.DelegatingSpliterator.OfInt((Supplier<Spliterator.OfInt>) supplier);
     }
@@ -190,6 +191,7 @@ abstract class IntPipeline<E_IN>
             Sink<Integer> opWrapSink(int flags, Sink<Long> sink) {
                 return new Sink.ChainedInt(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(int t) {
                         downstream.accept((long) t);
                     }
@@ -206,6 +208,7 @@ abstract class IntPipeline<E_IN>
             Sink<Integer> opWrapSink(int flags, Sink<Double> sink) {
                 return new Sink.ChainedInt(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(int t) {
                         downstream.accept((double) t);
                     }
@@ -245,6 +248,7 @@ abstract class IntPipeline<E_IN>
             Sink<Integer> opWrapSink(int flags, Sink<U> sink) {
                 return new Sink.ChainedInt(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(int t) {
                         downstream.accept(mapper.apply(t));
                     }

src/share/classes/java/util/stream/LongPipeline.java

@@ -148,6 +148,7 @@ abstract class LongPipeline<E_IN>
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     final Spliterator.OfLong lazySpliterator(Supplier<? extends Spliterator<Long>> supplier) {
         return new StreamSpliterators.DelegatingSpliterator.OfLong((Supplier<Spliterator.OfLong>) supplier);
     }
@@ -209,6 +210,7 @@ abstract class LongPipeline<E_IN>
             Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
                 return new Sink.ChainedLong(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(long t) {
                         downstream.accept(mapper.applyAsLong(t));
                     }
@@ -226,6 +228,7 @@ abstract class LongPipeline<E_IN>
             Sink<Long> opWrapSink(int flags, Sink<U> sink) {
                 return new Sink.ChainedLong(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(long t) {
                         downstream.accept(mapper.apply(t));
                     }
@@ -243,6 +246,7 @@ abstract class LongPipeline<E_IN>
             Sink<Long> opWrapSink(int flags, Sink<Integer> sink) {
                 return new Sink.ChainedLong(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(long t) {
                         downstream.accept(mapper.applyAsInt(t));
                     }

src/share/classes/java/util/stream/Nodes.java

@@ -60,6 +60,7 @@ final class Nodes {
      */
     static final long MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
 
+    @SuppressWarnings("raw")
     private static final Node EMPTY_NODE = new EmptyNode.OfRef();
     private static final Node.OfInt EMPTY_INT_NODE = new EmptyNode.OfInt();
     private static final Node.OfLong EMPTY_LONG_NODE = new EmptyNode.OfLong();
@@ -2150,7 +2151,7 @@ final class Nodes {
         }
 
         @Override
-        public void onCompletion(CountedCompleter caller) {
+        public void onCompletion(CountedCompleter<?> caller) {
             if (!isLeaf())
                 setLocalResult(concFactory.apply(leftChild.getLocalResult(), rightChild.getLocalResult()));
             super.onCompletion(caller);

src/share/classes/java/util/stream/ReduceOps.java

@@ -748,7 +748,7 @@ final class ReduceOps {
         }
 
         @Override
-        public void onCompletion(CountedCompleter caller) {
+        public void onCompletion(CountedCompleter<?> caller) {
             if (!isLeaf()) {
                 S leftResult = leftChild.getLocalResult();
                 leftResult.combine(rightChild.getLocalResult());

src/share/classes/java/util/stream/ReferencePipeline.java

@@ -170,9 +170,10 @@ abstract class ReferencePipeline<P_IN, P_OUT>
                     }
 
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(P_OUT u) {
                         if (predicate.test(u))
-                            downstream.accept(u);
+                            downstream.accept((Object) u);
                     }
                 };
             }
@@ -180,6 +181,7 @@ abstract class ReferencePipeline<P_IN, P_OUT>
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
         Objects.requireNonNull(mapper);
         return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
@@ -262,6 +264,7 @@ abstract class ReferencePipeline<P_IN, P_OUT>
                     }
 
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(P_OUT u) {
                         // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
                         Stream<? extends R> result = mapper.apply(u);
@@ -363,6 +366,7 @@ abstract class ReferencePipeline<P_IN, P_OUT>
             Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
                 return new Sink.ChainedReference<P_OUT>(sink) {
                     @Override
+                    @SuppressWarnings("unchecked")
                     public void accept(P_OUT u) {
                         tee.accept(u);
                         downstream.accept(u);
@@ -439,6 +443,7 @@ abstract class ReferencePipeline<P_IN, P_OUT>
         // The runtime type of U is never checked for equality with the component type of the runtime type of A[].
         // Runtime checking will be performed when an element is stored in A[], thus if A is not a
         // super type of U an ArrayStoreException will be thrown.
+        @SuppressWarnings("rawtypes")
         IntFunction rawGenerator = (IntFunction) generator;
         return (A[]) Nodes.flatten(evaluateToArrayNode(rawGenerator), rawGenerator)
                               .asArray(rawGenerator);

src/share/classes/java/util/stream/Sink.java

@@ -242,6 +242,7 @@ interface Sink<T> extends Consumer<T> {
      * {@code accept()} method on the downstream {@code Sink}.
      */
     static abstract class ChainedReference<T> implements Sink<T> {
+        @SuppressWarnings("rawtypes")
         protected final Sink downstream;
 
         public ChainedReference(Sink downstream) {
@@ -274,6 +275,7 @@ interface Sink<T> extends Consumer<T> {
      * {@code accept()} method on the downstream {@code Sink}.
      */
     static abstract class ChainedInt implements Sink.OfInt {
+        @SuppressWarnings("rawtypes")
         protected final Sink downstream;
 
         public ChainedInt(Sink downstream) {
@@ -306,6 +308,7 @@ interface Sink<T> extends Consumer<T> {
      * {@code accept()} method on the downstream {@code Sink}.
      */
     static abstract class ChainedLong implements Sink.OfLong {
+        @SuppressWarnings("rawtypes")
         protected final Sink downstream;
 
         public ChainedLong(Sink downstream) {
@@ -338,6 +341,7 @@ interface Sink<T> extends Consumer<T> {
      * {@code accept()} method on the downstream {@code Sink}.
      */
     static abstract class ChainedDouble implements Sink.OfDouble {
+        @SuppressWarnings("rawtypes")
         protected final Sink downstream;
 
         public ChainedDouble(Sink downstream) {

src/share/classes/java/util/stream/SortedOps.java

@@ -209,7 +209,7 @@ final class SortedOps {
         }
 
         @Override
-        public Sink<Long> opWrapSink(int flags, Sink sink) {
+        public Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
             Objects.requireNonNull(sink);
 
             if (StreamOpFlag.SORTED.isKnown(flags))
@@ -248,7 +248,7 @@ final class SortedOps {
         }
 
         @Override
-        public Sink<Double> opWrapSink(int flags, Sink sink) {
+        public Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
             Objects.requireNonNull(sink);
 
             if (StreamOpFlag.SORTED.isKnown(flags))
@@ -285,7 +285,7 @@ final class SortedOps {
         private T[] array;
         private int offset;
 
-        SizedRefSortingSink(Sink sink, Comparator<? super T> comparator) {
+        SizedRefSortingSink(Sink<T> sink, Comparator<? super T> comparator) {
             super(sink);
             this.comparator = comparator;
         }
@@ -324,7 +324,7 @@ final class SortedOps {
         private final Comparator<? super T> comparator;
         private ArrayList<T> list;
 
-        RefSortingSink(Sink sink, Comparator<? super T> comparator) {
+        RefSortingSink(Sink<T> sink, Comparator<? super T> comparator) {
             super(sink);
             this.comparator = comparator;
         }

src/share/classes/java/util/stream/StreamSpliterators.java

@@ -912,7 +912,8 @@ class StreamSpliterators {
             this.permits = new AtomicLong(limit >= 0 ? skip + limit : skip);
         }
 
-        UnorderedSliceSpliterator(T_SPLITR s, UnorderedSliceSpliterator parent) {
+        UnorderedSliceSpliterator(T_SPLITR s,
+                                  UnorderedSliceSpliterator<T, T_SPLITR> parent) {
             this.s = s;
             this.unlimited = parent.unlimited;
             this.permits = parent.permits;
@@ -991,7 +992,7 @@ class StreamSpliterators {
                 super(s, skip, limit);
             }
 
-            OfRef(Spliterator<T> s, OfRef parent) {
+            OfRef(Spliterator<T> s, OfRef<T> parent) {
                 super(s, parent);
             }