2.x: fix compilation errors when using Java 8 (#4893)

* Fix Java 8 compilation problems

* Suppress warnings for java 6

Author: akarnokd

src/main/java/io/reactivex/Flowable.java

@@ -1886,12 +1886,13 @@ public static <T> Flowable<T> fromFuture(Future<? extends T> future, long timeou
      * @return a Flowable that emits the item from the source {@link Future}
      * @see <a href="http://reactivex.io/documentation/operators/from.html">ReactiveX operators documentation: From</a>
      */
+    @SuppressWarnings({ "unchecked", "cast" })
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public static <T> Flowable<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit, Scheduler scheduler) {
         ObjectHelper.requireNonNull(scheduler, "scheduler is null");
-        return fromFuture(future, timeout, unit).subscribeOn(scheduler);
+        return fromFuture((Future<T>)future, timeout, unit).subscribeOn(scheduler);
     }
 
     /**
@@ -1923,12 +1924,13 @@ public static <T> Flowable<T> fromFuture(Future<? extends T> future, long timeou
      * @return a Flowable that emits the item from the source {@link Future}
      * @see <a href="http://reactivex.io/documentation/operators/from.html">ReactiveX operators documentation: From</a>
      */
+    @SuppressWarnings({ "cast", "unchecked" })
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public static <T> Flowable<T> fromFuture(Future<? extends T> future, Scheduler scheduler) {
         ObjectHelper.requireNonNull(scheduler, "scheduler is null");
-        return fromFuture(future).subscribeOn(scheduler);
+        return fromFuture((Future<T>)future).subscribeOn(scheduler);
     }
 
     /**
@@ -4145,13 +4147,14 @@ public static <T, R> Flowable<R> zip(Iterable<? extends Publisher<? extends T>>
      * @return a Flowable that emits the zipped results
      * @see <a href="http://reactivex.io/documentation/operators/zip.html">ReactiveX operators documentation: Zip</a>
      */
+    @SuppressWarnings({ "rawtypes", "unchecked", "cast" })
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T, R> Flowable<R> zip(Publisher<? extends Publisher<? extends T>> sources,
             final Function<? super Object[], ? extends R> zipper) {
         ObjectHelper.requireNonNull(zipper, "zipper is null");
-        return fromPublisher(sources).toList().flatMapPublisher(FlowableInternalHelper.<T, R>zipIterable(zipper));
+        return fromPublisher(sources).toList().flatMapPublisher((Function)FlowableInternalHelper.<T, R>zipIterable(zipper));
     }
 
     /**

src/test/java/io/reactivex/internal/operators/flowable/FlowableGroupByTest.java

@@ -1638,10 +1638,10 @@ public void accept(GroupedFlowable<Integer, Integer> g) {
     @Test
     public void keySelectorAndDelayError() {
         Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
-        .groupBy(Functions.identity(), true)
-        .flatMap(new Function<GroupedFlowable<Object, Integer>, Flowable<Integer>>() {
+        .groupBy(Functions.<Integer>identity(), true)
+        .flatMap(new Function<GroupedFlowable<Integer, Integer>, Flowable<Integer>>() {
             @Override
-            public Flowable<Integer> apply(GroupedFlowable<Object, Integer> g) throws Exception {
+            public Flowable<Integer> apply(GroupedFlowable<Integer, Integer> g) throws Exception {
                 return g;
             }
         })
@@ -1652,10 +1652,10 @@ public Flowable<Integer> apply(GroupedFlowable<Object, Integer> g) throws Except
     @Test
     public void keyAndValueSelectorAndDelayError() {
         Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
-        .groupBy(Functions.identity(), Functions.<Integer>identity(), true)
-        .flatMap(new Function<GroupedFlowable<Object, Integer>, Flowable<Integer>>() {
+        .groupBy(Functions.<Integer>identity(), Functions.<Integer>identity(), true)
+        .flatMap(new Function<GroupedFlowable<Integer, Integer>, Flowable<Integer>>() {
             @Override
-            public Flowable<Integer> apply(GroupedFlowable<Object, Integer> g) throws Exception {
+            public Flowable<Integer> apply(GroupedFlowable<Integer, Integer> g) throws Exception {
                 return g;
             }
         })

src/test/java/io/reactivex/internal/operators/observable/ObservableGroupByTest.java

@@ -1448,10 +1448,10 @@ public Integer apply(Integer i) {
     @Test
     public void keySelectorAndDelayError() {
         Observable.just(1).concatWith(Observable.<Integer>error(new TestException()))
-        .groupBy(Functions.identity(), true)
-        .flatMap(new Function<GroupedObservable<Object, Integer>, ObservableSource<Integer>>() {
+        .groupBy(Functions.<Integer>identity(), true)
+        .flatMap(new Function<GroupedObservable<Integer, Integer>, ObservableSource<Integer>>() {
             @Override
-            public ObservableSource<Integer> apply(GroupedObservable<Object, Integer> g) throws Exception {
+            public ObservableSource<Integer> apply(GroupedObservable<Integer, Integer> g) throws Exception {
                 return g;
             }
         })
@@ -1462,10 +1462,10 @@ public ObservableSource<Integer> apply(GroupedObservable<Object, Integer> g) thr
     @Test
     public void keyAndValueSelectorAndDelayError() {
         Observable.just(1).concatWith(Observable.<Integer>error(new TestException()))
-        .groupBy(Functions.identity(), Functions.<Integer>identity(), true)
-        .flatMap(new Function<GroupedObservable<Object, Integer>, ObservableSource<Integer>>() {
+        .groupBy(Functions.<Integer>identity(), Functions.<Integer>identity(), true)
+        .flatMap(new Function<GroupedObservable<Integer, Integer>, ObservableSource<Integer>>() {
             @Override
-            public ObservableSource<Integer> apply(GroupedObservable<Object, Integer> g) throws Exception {
+            public ObservableSource<Integer> apply(GroupedObservable<Integer, Integer> g) throws Exception {
                 return g;
             }
         })