@Test
	@Ignore //Fix or deprecate fromExecutor, this test might randomly hang on CI
	public void rejectedExecutionExceptionOnErrorSignalExecutor()
			throws InterruptedException {
		Exception exception = new IllegalStateException();
		final AtomicReference<Throwable> throwableInOnOperatorError =
				new AtomicReference<>();
		final AtomicReference<Object> dataInOnOperatorError = new AtomicReference<>();
		try {
			CountDownLatch hookLatch = new CountDownLatch(2);
			Hooks.onOperatorError((t, d) -> {
				throwableInOnOperatorError.set(t);
				dataInOnOperatorError.set(d);
				hookLatch.countDown();
				return t;
			});
			ExecutorService executor = newCachedThreadPool();
			CountDownLatch latch = new CountDownLatch(1);
			AssertSubscriber<Integer> assertSubscriber = new AssertSubscriber<>();
			Flux.range(0, 5)
			    .publishOn(fromExecutorService(executor))
			    .doOnNext(s -> {
				    try {
					    latch.await();
				    }
				    catch (InterruptedException e) {
					    throw Exceptions.propagate(exception);
				    }
			    })
			    .publishOn(fromExecutor(executor))
			    .subscribe(assertSubscriber);
			executor.shutdownNow();
			assertSubscriber.assertNoValues()
			                .assertNoError()
			                .assertNotComplete();
			hookLatch.await();
			assertThat(throwableInOnOperatorError.get()).isInstanceOf(RejectedExecutionException.class);
			assertThat(throwableInOnOperatorError.get().getSuppressed()[0]).isSameAs(exception);
		}
		finally {
			Hooks.resetOnOperatorError();
		}
	}
@Test
	@Ignore
	public void rejectedExecutionExceptionOnErrorSignalExecutor()
			throws InterruptedException {
		Exception exception = new IllegalStateException();
		final AtomicReference<Throwable> throwableInOnOperatorError =
				new AtomicReference<>();
		final AtomicReference<Object> dataInOnOperatorError = new AtomicReference<>();
		try {
			CountDownLatch hookLatch = new CountDownLatch(2);
			Hooks.onOperatorError((t, d) -> {
				throwableInOnOperatorError.set(t);
				dataInOnOperatorError.set(d);
				hookLatch.countDown();
				return t;
			});
			ExecutorService executor = newCachedThreadPool();
			CountDownLatch latch = new CountDownLatch(1);
			AssertSubscriber<Integer> assertSubscriber = new AssertSubscriber<>();
			Flux.range(0, 5)
			    .publishOn(fromExecutorService(executor))
			    .doOnNext(s -> {
				    try {
					    latch.await();
				    }
				    catch (InterruptedException e) {
					    throw Exceptions.propagate(exception);
				    }
			    })
			    .publishOn(fromExecutor(executor))
			    .subscribe(assertSubscriber);
			executor.shutdownNow();
			assertSubscriber.assertNoValues()
			                .assertNoError()
			                .assertNotComplete();
			hookLatch.await();
			assertThat(throwableInOnOperatorError.get()).isInstanceOf(RejectedExecutionException.class);
			assertThat(throwableInOnOperatorError.get().getSuppressed()[0]).isSameAs(exception);
		}
		finally {
			Hooks.resetOnOperatorError();
		}
	}
@test @ignore public void rejectedexecutionexceptiononerrorsignalexecutor() throws interruptedexception { exception exception = new illegalstateexception(); final atomicreference<throwable> throwableinonoperatorerror = new atomicreference<>(); final atomicreference<object> datainonoperatorerror = new atomicreference<>(); try { countdownlatch hooklatch = new countdownlatch(2); hooks.onoperatorerror((t, d) -> { throwableinonoperatorerror.set(t); datainonoperatorerror.set(d); hooklatch.countdown(); return t; }); executorservice executor = newcachedthreadpool(); countdownlatch latch = new countdownlatch(1); assertsubscriber<integer> assertsubscriber = new assertsubscriber<>(); flux.range(0, 5) .publishon(fromexecutorservice(executor)) .doonnext(s -> { try { latch.await(); } catch (interruptedexception e) { throw exceptions.propagate(exception); } }) .publishon(fromexecutor(executor)) .subscribe(assertsubscriber); executor.shutdownnow(); assertsubscriber.assertnovalues() .assertnoerror() .assertnotcomplete(); hooklatch.await(); assertthat(throwableinonoperatorerror.get()).isinstanceof(rejectedexecutionexception.class); assertthat(throwableinonoperatorerror.get().getsuppressed()[0]).issameas(exception); } finally { hooks.resetonoperatorerror(); } }