Default methods that return async sources should not throw

Motivation:

apple#2168 and apple#2235 added default implementations that throw an exception.
However, those method signatures return an async source.

Modifications:

- Return a `failed` source with an exception instead of throwing;

Result:

Default methods follow async contract.

servicetalk-client-api/src/main/java/io/servicetalk/client/api/ConnectionFactory.java

@@ -24,6 +24,7 @@
 import javax.annotation.Nullable;
 
 import static io.servicetalk.client.api.DeprecatedToNewConnectionFactoryFilter.CONNECTION_FACTORY_CONTEXT_MAP_KEY;
+import static io.servicetalk.concurrent.api.Single.failed;
 
 /**
  * A factory for creating new connections.
@@ -46,8 +47,9 @@ public interface ConnectionFactory<ResolvedAddress, C extends ListenableAsyncClo
      */
     @Deprecated // FIXME: 0.43 - remove deprecated method
     default Single<C> newConnection(ResolvedAddress address, @Nullable TransportObserver observer) {
-        throw new UnsupportedOperationException("ConnectionFactory#newConnection(ResolvedAddress, TransportObserver) " +
-                "is not supported by " + getClass());
+        return failed(new UnsupportedOperationException(
+                "ConnectionFactory#newConnection(ResolvedAddress, TransportObserver) is not supported by " +
+                        getClass()));
     }
 
     /**

servicetalk-client-api/src/main/java/io/servicetalk/client/api/LoadBalancer.java

@@ -25,6 +25,8 @@
 import java.util.function.Predicate;
 import javax.annotation.Nullable;
 
+import static io.servicetalk.concurrent.api.Single.failed;
+
 /**
  * Given multiple {@link SocketAddress}es select the most desired {@link SocketAddress} to use. This is typically used
  * to determine which connection to issue a request to.
@@ -48,8 +50,8 @@ public interface LoadBalancer<C extends LoadBalancedConnection> extends Listenab
      */
     @Deprecated
     default Single<C> selectConnection(Predicate<C> selector) { // FIXME: 0.43 - remove deprecated method
-        throw new UnsupportedOperationException("LoadBalancer#selectConnection(Predicate) is not supported by " +
-                getClass());
+        return failed(new UnsupportedOperationException(
+                "LoadBalancer#selectConnection(Predicate) is not supported by " + getClass()));
     }
 
     /**

servicetalk-grpc-protoc/src/main/java/io/servicetalk/grpc/protoc/Generator.java

@@ -899,9 +899,11 @@ private TypeSpec.Builder addClientInterfaces(final State state, final TypeSpec.B
                                     .addJavadoc(JAVADOC_PARAM + metadata +
                                             " the metadata associated with this client call." + lineSeparator());
                                 }
-                                return b.addStatement("throw new UnsupportedOperationException(\"This method is not " +
-                                        "implemented by \" + getClass() + \". Consider migrating to an alternative " +
-                                        "method or implement this method if it's required temporarily.\")");
+                                return b.addStatement("return $T.failed(new UnsupportedOperationException(\"" +
+                                        "This method is not implemented by \" + getClass() + \". Consider migrating " +
+                                        "to an alternative method or implement this method if it's required " +
+                                        "temporarily.\"))", clientMetaData.methodProto.getServerStreaming() ?
+                                        Publisher : Single);
                             }))
                     .addMethod(newRpcMethodSpec(clientMetaData.methodProto, EnumSet.of(INTERFACE, CLIENT),
                             printJavaDocs, (methodName, b) -> {
@@ -1050,7 +1052,7 @@ private MethodSpec newRpcMethodSpec(
 
     private static MethodSpec newRpcMethodSpec(
             final ClassName inClass, final ClassName outClass, final String methodName,
-            final boolean clientSteaming, final boolean serverStreaming,
+            final boolean clientStreaming, final boolean serverStreaming,
             final EnumSet<NewRpcMethodFlag> flags, final boolean printJavaDocs,
             final BiFunction<String, MethodSpec.Builder, MethodSpec.Builder> methodBuilderCustomizer) {
         final MethodSpec.Builder methodSpecBuilder = methodBuilderCustomizer.apply(methodName,
@@ -1059,7 +1061,7 @@ private static MethodSpec newRpcMethodSpec(
         final Modifier[] mods = flags.contains(INTERFACE) ? new Modifier[0] : new Modifier[]{FINAL};
 
         if (flags.contains(BLOCKING)) {
-            if (clientSteaming) {
+            if (clientStreaming) {
                 if (flags.contains(CLIENT)) {
                     methodSpecBuilder.addParameter(ParameterizedTypeName.get(Types.Iterable, inClass), request, mods);
                     if (printJavaDocs) {
@@ -1116,7 +1118,7 @@ private static MethodSpec newRpcMethodSpec(
                                 "propagated to the peer.", GrpcStatusException);
             }
         } else {
-            if (clientSteaming) {
+            if (clientStreaming) {
                 methodSpecBuilder.addParameter(ParameterizedTypeName.get(Publisher, inClass), request, mods);
                 if (printJavaDocs) {
                     methodSpecBuilder.addJavadoc(JAVADOC_PARAM + request +