{"action":"created","issue":{"url":"https://api.github.com/repos/reactor/reactor-core/issues/1107","repository_url":"https://api.github.com/repos/reactor/reactor-core","labels_url":"https://api.github.com/repos/reactor/reactor-core/issues/1107/labels{/name}","comments_url":"https://api.github.com/repos/reactor/reactor-core/issues/1107/comments","events_url":"https://api.github.com/repos/reactor/reactor-core/issues/1107/events","html_url":"https://github.com/reactor/reactor-core/issues/1107","id":301741398,"number":1107,"title":"Premature thread interruption when using Mono.just().transform(flux)","user":{"login":"dfeist","id":380765,"avatar_url":"https://avatars3.githubusercontent.com/u/380765?v=4","gravatar_id":"","url":"https://api.github.com/users/dfeist","html_url":"https://github.com/dfeist","followers_url":"https://api.github.com/users/dfeist/followers","following_url":"https://api.github.com/users/dfeist/following{/other_user}","gists_url":"https://api.github.com/users/dfeist/gists{/gist_id}","starred_url":"https://api.github.com/users/dfeist/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dfeist/subscriptions","organizations_url":"https://api.github.com/users/dfeist/orgs","repos_url":"https://api.github.com/users/dfeist/repos","events_url":"https://api.github.com/users/dfeist/events{/privacy}","received_events_url":"https://api.github.com/users/dfeist/received_events","type":"User","site_admin":false},"labels":[{"id":284719891,"url":"https://api.github.com/repos/reactor/reactor-core/labels/enhancement","name":"enhancement","color":"84b6eb","default":true},{"id":710296502,"url":"https://api.github.com/repos/reactor/reactor-core/labels/has-workaround","name":"has-workaround","color":"159818","default":false}],"state":"open","locked":false,"assignee":{"login":"simonbasle","id":6986166,"avatar_url":"https://avatars0.githubusercontent.com/u/6986166?v=4","gravatar_id":"","url":"https://api.github.com/users/simonbasle","html_url":"https://github.com/simonbasle","followers_url":"https://api.github.com/users/simonbasle/followers","following_url":"https://api.github.com/users/simonbasle/following{/other_user}","gists_url":"https://api.github.com/users/simonbasle/gists{/gist_id}","starred_url":"https://api.github.com/users/simonbasle/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/simonbasle/subscriptions","organizations_url":"https://api.github.com/users/simonbasle/orgs","repos_url":"https://api.github.com/users/simonbasle/repos","events_url":"https://api.github.com/users/simonbasle/events{/privacy}","received_events_url":"https://api.github.com/users/simonbasle/received_events","type":"User","site_admin":false},"assignees":[{"login":"simonbasle","id":6986166,"avatar_url":"https://avatars0.githubusercontent.com/u/6986166?v=4","gravatar_id":"","url":"https://api.github.com/users/simonbasle","html_url":"https://github.com/simonbasle","followers_url":"https://api.github.com/users/simonbasle/followers","following_url":"https://api.github.com/users/simonbasle/following{/other_user}","gists_url":"https://api.github.com/users/simonbasle/gists{/gist_id}","starred_url":"https://api.github.com/users/simonbasle/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/simonbasle/subscriptions","organizations_url":"https://api.github.com/users/simonbasle/orgs","repos_url":"https://api.github.com/users/simonbasle/repos","events_url":"https://api.github.com/users/simonbasle/events{/privacy}","received_events_url":"https://api.github.com/users/simonbasle/received_events","type":"User","site_admin":false}],"milestone":{"url":"https://api.github.com/repos/reactor/reactor-core/milestones/18","html_url":"https://github.com/reactor/reactor-core/milestone/18","labels_url":"https://api.github.com/repos/reactor/reactor-core/milestones/18/labels","id":2516193,"number":18,"title":"3.2.0.RELEASE","description":"","creator":{"login":"smaldini","id":76718,"avatar_url":"https://avatars2.githubusercontent.com/u/76718?v=4","gravatar_id":"","url":"https://api.github.com/users/smaldini","html_url":"https://github.com/smaldini","followers_url":"https://api.github.com/users/smaldini/followers","following_url":"https://api.github.com/users/smaldini/following{/other_user}","gists_url":"https://api.github.com/users/smaldini/gists{/gist_id}","starred_url":"https://api.github.com/users/smaldini/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/smaldini/subscriptions","organizations_url":"https://api.github.com/users/smaldini/orgs","repos_url":"https://api.github.com/users/smaldini/repos","events_url":"https://api.github.com/users/smaldini/events{/privacy}","received_events_url":"https://api.github.com/users/smaldini/received_events","type":"User","site_admin":false},"open_issues":38,"closed_issues":14,"state":"open","created_at":"2017-05-11T21:19:24Z","updated_at":"2018-03-02T15:02:11Z","due_on":null,"closed_at":null},"comments":1,"created_at":"2018-03-02T11:48:54Z","updated_at":"2018-03-02T15:08:23Z","closed_at":null,"author_association":"CONTRIBUTOR","body":"See: https://github.com/ReactiveX/RxJava/issues/5711\r\nSee: https://github.com/ReactiveX/RxJava/issues/5715\r\n\r\nIn the following scenario:\r\n```\r\nScheduler scheduler = Schedulers.fromExecutorService(Executors.newCachedThreadPool());\r\nFlux<Integer> flux = Flux.just(1);\r\nFunction<Publisher<Integer>, Publisher<Integer>> addOne = f -> Flux.from(f)\r\n\t\t                                                    .map(i -> i + 1)\r\n\t\t                                                    .publishOn(scheduler);\r\nflux.concatMap(j -> Mono.just(j).transform(addOne)).subscribe(System.out::println);\r\n````\r\n\r\n- the Mono.transform, given addOne is not a Mono uses MonoNext to wrap\r\n- MonoNext does s.cancel() before actual.onNext(t);\r\n- WorkerTask future gets cancelled (due to MonoNext s.cancel), and can get cancelled even before reactor.core.scheduler.WorkerTask#setFuture. If this happens thread is interrupted and processing on that thread (e.g. reading of IO) can be cut short\r\n\r\nIn my case the thread doing the map() is an HTTP selector thread and the `Thread.interrupt()` provoked in `WorkerTask#setFuture` causes the reading of `AbstractInterruptibleChannel` to be cut short with a `ClosedByInterruptException` resulting in an empty response body.\r\n\r\nThe cancel-before-next in `MonoNext` is necessary but causes a race condition between `reactor.core.scheduler.WorkerTask#dispose` and `sr.setFuture(f)` (in `reactor.core.scheduler.Schedulers#workerSchedule`), where when `dispose()` is called before `setFuture(f)` the thread that is still processing the `onNext()` signal is prematurely interrupted.\r\n"},"comment":{"url":"https://api.github.com/repos/reactor/reactor-core/issues/comments/369946187","html_url":"https://github.com/reactor/reactor-core/issues/1107#issuecomment-369946187","issue_url":"https://api.github.com/repos/reactor/reactor-core/issues/1107","id":369946187,"user":{"login":"simonbasle","id":6986166,"avatar_url":"https://avatars0.githubusercontent.com/u/6986166?v=4","gravatar_id":"","url":"https://api.github.com/users/simonbasle","html_url":"https://github.com/simonbasle","followers_url":"https://api.github.com/users/simonbasle/followers","following_url":"https://api.github.com/users/simonbasle/following{/other_user}","gists_url":"https://api.github.com/users/simonbasle/gists{/gist_id}","starred_url":"https://api.github.com/users/simonbasle/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/simonbasle/subscriptions","organizations_url":"https://api.github.com/users/simonbasle/orgs","repos_url":"https://api.github.com/users/simonbasle/repos","events_url":"https://api.github.com/users/simonbasle/events{/privacy}","received_events_url":"https://api.github.com/users/simonbasle/received_events","type":"User","site_admin":false},"created_at":"2018-03-02T15:08:22Z","updated_at":"2018-03-02T15:08:22Z","author_association":"CONTRIBUTOR","body":"`has-workaround`: The particular code snippet can be changed to using `Flux.just(f)` in the `concatMap` instead of `Mono.just(j)`, which will avoid the `MonoNext` and associated cancellation altogether."}}
{"actor":{"display_login":"simonbasle"}}