SerializedProcessor extends FlowableProcessor<T> { @Override public void onError(Throwable t) { if (done) { RxJavaPlugins.onError(t); return; } boolean reportError; synchronized (this) { if (done) { reportError = true; } else { done = true; if (emitting) { AppendOnlyLinkedArrayList<Object> q = queue; if (q == null) { q = new AppendOnlyLinkedArrayList<Object>(4); queue = q; } q.setFirst(NotificationLite.error(t)); return; } reportError = false; emitting = true; } } if (reportError) { RxJavaPlugins.onError(t); return; } actual.onError(t); } SerializedProcessor(final FlowableProcessor<T> actual); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override boolean hasSubscribers(); @Override boolean hasThrowable(); @Override @Nullable Throwable getThrowable(); @Override boolean hasComplete();  }
@Test public void onErrorOnErrorRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final FlowableProcessor<Integer> s = PublishProcessor.<Integer>create().toSerialized(); TestSubscriber<Integer> ts = s.test(); final TestException ex = new TestException(); List<Throwable> errors = TestHelper.trackPluginErrors(); try { Runnable r1 = new Runnable() { @Override public void run() { s.onError(ex); } }; Runnable r2 = new Runnable() { @Override public void run() { s.onError(ex); } }; TestHelper.race(r1, r2); ts.assertFailure(TestException.class); TestHelper.assertUndeliverable(errors, 0, TestException.class); } finally { RxJavaPlugins.reset(); } } }