{"record":{"id":"3661be282e82fe39","repo":"ReactiveX/RxJava","slug":"the-source-did-not-signal-an-event-for-timeout","errorCode":null,"errorMessage":"The source did not signal an event for {timeout} {unit} and has been terminated.","messagePattern":"The source did not signal an event for (.+?) (.+?) and has been terminated\\.","errorType":"exception","errorClass":"TimeoutException","httpStatus":null,"severity":"error","filePath":"src/main/java/io/reactivex/rxjava4/internal/observers/FutureMultiObserver.java","lineNumber":97,"sourceCode":"            await();\n        }\n\n        if (isCancelled()) {\n            throw new CancellationException();\n        }\n        Throwable ex = error;\n        if (ex != null) {\n            throw new ExecutionException(ex);\n        }\n        return value;\n    }\n\n    @Override\n    public T get(long timeout, @NonNull TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {\n        if (getCount() != 0) {\n            BlockingHelper.verifyNonBlocking();\n            if (!await(timeout, unit)) {\n                throw new TimeoutException(timeoutMessage(timeout, unit));\n            }\n        }\n\n        if (isCancelled()) {\n            throw new CancellationException();\n        }\n\n        Throwable ex = error;\n        if (ex != null) {\n            throw new ExecutionException(ex);\n        }\n        return value;\n    }\n\n    @Override\n    public void onSubscribe(Disposable d) {\n        DisposableHelper.setOnce(this.upstream, d);\n    }","sourceCodeStart":79,"sourceCodeEnd":115,"githubUrl":"https://github.com/ReactiveX/RxJava/blob/a8ab5356143f37a8f1dd6d76c79191bec5fa343b/src/main/java/io/reactivex/rxjava4/internal/observers/FutureMultiObserver.java#L79-L115","documentation":"Thrown as a TimeoutException by FutureMultiObserver.get(long, TimeUnit) when the backing source does not signal (onNext/onError/onComplete) within the given timeout. The message is built by ExceptionHelper.timeoutMessage and reads 'The source did not signal an event for {timeout} {unit} and has been terminated.' — note 'has been terminated' here means the get() call gave up, not necessarily that the source itself errored. FutureMultiObserver implements Future<T> over a Multi source.","triggerScenarios":"Calling futureMulti.get(5, TimeUnit.SECONDS) where the upstream Multi neither emits, errors, nor completes within 5 seconds. Common with cold sources that never complete (e.g. infinite intervals), sources blocked on I/O, or sources awaiting a signal that never arrives.","commonSituations":"Blocking on a Multi that wraps a never-completing stream; network/datasource stalls; deadlocks where the producer thread is starved; timeouts set too low for the workload; forgetting that BlockingHelper.verifyNonBlocking() also runs and may throw if called from a non-blocking context.","solutions":["Increase the timeout to match realistic worst-case latency.","Ensure the upstream Multi actually terminates — apply a timeout() operator on the source so it errors/completes deterministically instead of hanging the Future.","Catch TimeoutException explicitly and handle it (retry, fallback, or report).","Verify you are calling get(timeout,unit) from a thread permitted to block (BlockingHelper rejects blocking on certain schedulers)."],"exampleFix":"// before\nMulti<T> m = source; // may never complete\nT v = m.toFuture().get(2, TimeUnit.SECONDS); // TimeoutException\n\n// after\nMulti<T> m = source.timeout(Duration.ofSeconds(2));\nT v;\ntry {\n    v = m.toFuture().get(5, TimeUnit.SECONDS);\n} catch (TimeoutException e) {\n    v = fallback();\n}","handlingStrategy":"try-catch","validationCode":"// choose a timeout that covers realistic latency, then also bound the source:\nMulti<T> bounded = source.timeout(Duration.ofSeconds(5));\n// FutureMultiObserver.get declares TimeoutException, so you must handle it anyway.","typeGuard":null,"tryCatchPattern":"try {\n    T v = multi.toFuture().get(5, TimeUnit.SECONDS);\n} catch (TimeoutException e) {\n    v = fallback();\n} catch (InterruptedException | ExecutionException e) {\n    Thread.currentThread().interrupt(); // for InterruptedException\n    throw new RuntimeException(e);\n}","preventionTips":["Always apply a timeout() operator on sources feeding a Future so they terminate deterministically.","Size get(timeout) to worst-case latency, not average.","Confirm the calling thread is allowed to block (BlockingHelper may reject it)."],"tags":["timeout","future","multi","blocking","rxjava"],"backgroundTag":null,"analyzedSha":"a8ab5356143f37a8f1dd6d76c79191bec5fa343b","analyzedAt":"2026-08-13T23:25:30.069Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}