fix unchecked casts

src/main/java/com/google/api/gax/core/ApiFutures.java

@@ -59,7 +59,7 @@ public void onSuccess(V v) {
         });
   }
 
-  public static <V, X extends Throwable> ApiFuture catching(
+  public static <V, X extends Throwable> ApiFuture<V> catching(
       ApiFuture<? extends V> input,
       Class<X> exceptionType,
       ApiFunction<? super X, ? extends V> callback) {
@@ -68,7 +68,7 @@ public static <V, X extends Throwable> ApiFuture catching(
             listenableFutureForApiFuture(input),
             exceptionType,
             new GaxFunctionToGuavaFunction<X, V>(callback));
-    return new ListenableFutureToApiFuture(catchingFuture);
+    return new ListenableFutureToApiFuture<V>(catchingFuture);
   }
 
   public static <V> ApiFuture<V> immediateFuture(V value) {
@@ -90,9 +90,9 @@ private static <V> ListenableFuture<V> listenableFutureForApiFuture(ApiFuture<V>
     ListenableFuture<V> listenableFuture;
     if (apiFuture instanceof AbstractApiFuture) {
       // prefer to use the wrapped ListenableFuture to reduce the number of layers
-      listenableFuture = ((AbstractApiFuture) apiFuture).getInternalListenableFuture();
+      listenableFuture = ((AbstractApiFuture<V>) apiFuture).getInternalListenableFuture();
     } else {
-      listenableFuture = new ApiFutureToListenableFuture(apiFuture);
+      listenableFuture = new ApiFutureToListenableFuture<V>(apiFuture);
     }
     return listenableFuture;
   }

src/main/java/com/google/api/gax/core/SettableApiFuture.java

@@ -37,7 +37,7 @@ public final class SettableApiFuture<V> extends AbstractApiFuture<V> {
 
   private SettableApiFuture() {}
 
-  public static SettableApiFuture create() {
+  public static <V> SettableApiFuture<V> create() {
     return new SettableApiFuture<>();
   }
 }

src/main/java/com/google/api/gax/grpc/DirectStreamingCallable.java

@@ -62,7 +62,7 @@ void serverStreamingCall(
     ClientCall<RequestT, ResponseT> call =
         factory.newCall(context.getChannel(), context.getCallOptions());
     ClientCalls.asyncServerStreamingCall(
-        call, request, new RpcStreamObserverDelegate(responseObserver));
+        call, request, new RpcStreamObserverDelegate<ResponseT>(responseObserver));
   }
 
   Iterator<ResponseT> blockingServerStreamingCall(RequestT request, CallContext context) {
@@ -77,18 +77,19 @@ ApiStreamObserver<RequestT> bidiStreamingCall(
     Preconditions.checkNotNull(responseObserver);
     ClientCall<RequestT, ResponseT> call =
         factory.newCall(context.getChannel(), context.getCallOptions());
-    return new StreamObserverDelegate(
-        ClientCalls.asyncBidiStreamingCall(call, new RpcStreamObserverDelegate(responseObserver)));
+    return new StreamObserverDelegate<RequestT>(
+        ClientCalls.asyncBidiStreamingCall(
+            call, new RpcStreamObserverDelegate<ResponseT>(responseObserver)));
   }
 
   ApiStreamObserver<RequestT> clientStreamingCall(
       ApiStreamObserver<ResponseT> responseObserver, CallContext context) {
     Preconditions.checkNotNull(responseObserver);
     ClientCall<RequestT, ResponseT> call =
         factory.newCall(context.getChannel(), context.getCallOptions());
-    return new StreamObserverDelegate(
+    return new StreamObserverDelegate<RequestT>(
         ClientCalls.asyncClientStreamingCall(
-            call, new RpcStreamObserverDelegate(responseObserver)));
+            call, new RpcStreamObserverDelegate<ResponseT>(responseObserver)));
   }
 
   private static class RpcStreamObserverDelegate<V> implements StreamObserver<V> {