Fix a couple of lint and other generics issues

Author: akarnokd

build.gradle

@@ -65,6 +65,8 @@ java {
 
 tasks.withType(JavaCompile) {
     options.compilerArgs << "-parameters"
+    options.compilerArgs += '-Xlint:-module'
+    options.compilerArgs += '-Xlint:unchecked'
 }
 
 apply from: file("gradle/javadoc_cleanup.gradle")

src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java

@@ -16,6 +16,7 @@
 import java.lang.Thread.UncaughtExceptionHandler;
 import java.util.Objects;
 import java.util.concurrent.*;
+import java.util.concurrent.Flow.Subscriber;
 
 import static java.util.concurrent.Flow.*;
 
@@ -115,7 +116,7 @@ public final class RxJavaPlugins {
 
     @SuppressWarnings("rawtypes")
     @Nullable
-    static volatile BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber[], @NonNull ? extends Subscriber[]> onParallelSubscribe;
+    static volatile BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber<@NonNull ?>[], @NonNull ? extends Subscriber<@NonNull ?>[]> onParallelSubscribe;
 
     @Nullable
     static volatile BooleanSupplier onBeforeBlocking;
@@ -1008,12 +1009,12 @@ public static CompletableObserver onSubscribe(@NonNull Completable source, @NonN
      * @param subscribers the array of subscribers
      * @return the value returned by the hook
      */
-    @SuppressWarnings({ "rawtypes" })
+    @SuppressWarnings({ "unchecked" })
     @NonNull
-    public static <@NonNull T> Subscriber<? super T>[] onSubscribe(@NonNull ParallelFlowable<T> source, @NonNull Subscriber<? super T>[] subscribers) {
-        BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber[], @NonNull ? extends Subscriber[]> f = onParallelSubscribe;
+    public static <@NonNull T> Subscriber<? super T>[] onSubscribe(@NonNull ParallelFlowable<? extends T> source, @NonNull Subscriber<? super T>[] subscribers) {
+        var f = onParallelSubscribe;
         if (f != null) {
-            return apply(f, source, subscribers);
+            return (@NonNull Subscriber<@NonNull ? super @NonNull T>[]) apply(f, source, subscribers);
         }
         return subscribers;
     }
@@ -1161,7 +1162,7 @@ public static void setOnParallelAssembly(@Nullable Function<? super ParallelFlow
      * @since 3.1.0
      */
     @SuppressWarnings("rawtypes")
-    public static void setOnParallelSubscribe(@Nullable BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber[], @NonNull ? extends Subscriber[]> handler) {
+    public static void setOnParallelSubscribe(@Nullable BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber<@NonNull ?>[], @NonNull ? extends Subscriber<@NonNull ?>[]> handler) {
         if (lockdown) {
             throw new IllegalStateException("Plugins can't be changed anymore");
         }
@@ -1176,7 +1177,7 @@ public static void setOnParallelSubscribe(@Nullable BiFunction<? super ParallelF
      */
     @SuppressWarnings("rawtypes")
     @Nullable
-    public static BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber[], @NonNull ? extends Subscriber[]> getOnParallelSubscribe() {
+    public static BiFunction<? super ParallelFlowable, @NonNull ? super Subscriber<@NonNull ?>[], @NonNull ? extends Subscriber<@NonNull ?>[]> getOnParallelSubscribe() {
         return onParallelSubscribe;
     }
 
@@ -1355,7 +1356,7 @@ public static Scheduler createExecutorScheduler(@NonNull Executor executor, bool
      * @return the result of the function call
      */
     @NonNull
-    static <@NonNull T, @NonNull U, @NonNull R> R apply(@NonNull BiFunction<T, U, R> f, @NonNull T t, @NonNull U u) {
+    static <@NonNull T, @NonNull U, @NonNull R> R apply(@NonNull BiFunction<? super T, ? super U, ? extends R> f, @NonNull T t, @NonNull U u) {
         try {
             return f.apply(t, u);
         } catch (Throwable ex) {

src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java

@@ -589,14 +589,12 @@ public void onComplete() {
         .assertComplete();
     }
 
-    @SuppressWarnings("rawtypes")
     @Test
     public void parallelFlowableStart() {
         try {
-            RxJavaPlugins.setOnParallelSubscribe(new BiFunction<ParallelFlowable, Subscriber[], Subscriber[]>() {
-                @Override
-                public Subscriber[] apply(ParallelFlowable f, final Subscriber[] t) {
-                    return new Subscriber[] { new Subscriber() {
+            RxJavaPlugins.setOnParallelSubscribe((_, t) -> {
+                    var result = new Subscriber<?>[] {
+                        new Subscriber<Object>() {
 
                             @Override
                             public void onSubscribe(Subscription s) {
@@ -606,7 +604,7 @@ public void onSubscribe(Subscription s) {
                             @SuppressWarnings("unchecked")
                             @Override
                             public void onNext(Object value) {
-                                t[0].onNext((Integer)value - 9);
+                                ((Subscriber<Integer>)t[0]).onNext(((Integer)value - 9));
                             }
 
                             @Override
@@ -621,8 +619,9 @@ public void onComplete() {
 
                         }
                     };
+                    return result;
                 }
-            });
+            );
 
             Flowable.range(10, 3)
             .parallel(1)