fetcher.thread.timeout (#996, #1861) wraps protocol.getProtocolOutput() in a Future run on a single-thread executor owned by each FetcherThread, and cancels it with future.cancel(true) when the deadline passes.
Two problems with that:
cancel(true) is a Thread.interrupt(), which okhttp does not honour while connecting or reading. The helper thread stays blocked on the socket until okhttp's own timeouts fire, and the FetcherThread queues behind it at its next fetch because the executor is single-threaded. One host that dribbles bytes can take a FetcherThread out of service for as long as http.timeout, or topology.message.timeout.secs (300s by default) which is the value the okhttp callTimeout is currently set from.- It doubles the thread count of the bolt whenever the option is on: 50
FetcherThreads plus 50 FetcherTimeout helpers.
okhttp exposes the right primitive: Call.timeout() sets a per-call deadline enforced by okio's single shared watchdog thread, and on expiry the call is cancelled, the socket closed, and the fetching thread gets an InterruptedIOException immediately.
Proposal: okhttp.HttpProtocol applies fetcher.thread.timeout as a per-call deadline, and Protocol gains a default boolean supportsFetchTimeout() (false by default, true for okhttp when configured). FetcherBolt and SimpleFetcherBolt keep the Future path only for protocols that do not support it, creating the helper lazily. For the default okhttp protocol the fetch runs in the FetcherThread itself, the timeout is a real cancellation, and no helper threads exist.
fetcher.thread.timeout(#996, #1861) wrapsprotocol.getProtocolOutput()in aFuturerun on a single-thread executor owned by eachFetcherThread, and cancels it withfuture.cancel(true)when the deadline passes.Two problems with that:
cancel(true)is aThread.interrupt(), which okhttp does not honour while connecting or reading. The helper thread stays blocked on the socket until okhttp's own timeouts fire, and theFetcherThreadqueues behind it at its next fetch because the executor is single-threaded. One host that dribbles bytes can take aFetcherThreadout of service for as long ashttp.timeout, ortopology.message.timeout.secs(300s by default) which is the value the okhttpcallTimeoutis currently set from.FetcherThreads plus 50FetcherTimeouthelpers.okhttp exposes the right primitive:
Call.timeout()sets a per-call deadline enforced by okio's single shared watchdog thread, and on expiry the call is cancelled, the socket closed, and the fetching thread gets anInterruptedIOExceptionimmediately.Proposal:
okhttp.HttpProtocolappliesfetcher.thread.timeoutas a per-call deadline, andProtocolgains adefault boolean supportsFetchTimeout()(false by default, true for okhttp when configured).FetcherBoltandSimpleFetcherBoltkeep theFuturepath only for protocols that do not support it, creating the helper lazily. For the default okhttp protocol the fetch runs in theFetcherThreaditself, the timeout is a real cancellation, and no helper threads exist.