Publisher#flatMapConcatIterable propagate error even if no onSubscribe (#1671)

Motivation:
The ReactiveStreams specification says that onSubscribe(Subscription)
must be called prior to any other signals being delivered [1]. However
in practice user code may throw in onSubscribe which may stop the
propagation. In this case Publisher#flatMapConcatIterable has an
assertion which may prevent the error from propagating downstream and
result in a control flow deadlock.

[1] https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/README.md#1.9

Modifications:
- Move assertion in Publisher#flatMapConcatIterable to allow for null
  subscription, add a runtime check before it is used.

Result:
More robust error propagation for Publisher#flatMapConcatIterable in the
precense of Reactive Streams specification violations.

Author: Scottmitch

servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/PublisherConcatMapIterable.java

@@ -172,7 +172,6 @@ private enum ErrorHandlingStrategyInDrain {
         }
 
         private void tryDrainIterator(ErrorHandlingStrategyInDrain errorHandlingStrategyInDrain) {
-            assert sourceSubscription != null;
             boolean hasNext = false;
             boolean thrown = false;
             boolean terminated = false;
@@ -241,7 +240,9 @@ private void tryDrainIterator(ErrorHandlingStrategyInDrain errorHandlingStrategy
                                 // here before we unlock emitting so visibility to other threads should be taken care of
                                 // by the write to emitting below (and later read).
                                 currentIterator = EmptyIterator.instance();
-                                sourceSubscription.request(1);
+                                if (sourceSubscription != null) {
+                                    sourceSubscription.request(1);
+                                }
                             }
                         } finally {
                             // The lock must be released after we interact with the subscription for thread safety

servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/PublisherConcatMapIterableTest.java

@@ -68,6 +68,25 @@ class PublisherConcatMapIterableTest {
     private final TestPublisherSubscriber<String> subscriber = new TestPublisherSubscriber<>();
     private final TestSubscription subscription = new TestSubscription();
 
+    @Test
+    void onSubscribeThrowsPropagatesError() throws Exception {
+        @SuppressWarnings("unchecked")
+        Subscriber<String> mockSubscriber = mock(Subscriber.class);
+        CountDownLatch latchOnError = new CountDownLatch(1);
+        AtomicReference<Throwable> causeRef = new AtomicReference<>();
+        doAnswer(a -> {
+            causeRef.set(a.getArgument(0));
+            latchOnError.countDown();
+            return null;
+        }).when(mockSubscriber).onError(any());
+
+        toSource(from(singletonList("foo")).beforeOnSubscribe(subscription1 -> {
+            throw DELIBERATE_EXCEPTION;
+        }).flatMapConcatIterable(identity())).subscribe(mockSubscriber);
+        latchOnError.await();
+        assertThat(causeRef.get(), is(DELIBERATE_EXCEPTION));
+    }
+
     @Test
     void upstreamRecoverWithMakesProgress() throws Exception {
         @SuppressWarnings("unchecked")