{"action":"created","issue":{"url":"https://api.github.com/repos/spring-projects/spring-amqp/issues/432","repository_url":"https://api.github.com/repos/spring-projects/spring-amqp","labels_url":"https://api.github.com/repos/spring-projects/spring-amqp/issues/432/labels{/name}","comments_url":"https://api.github.com/repos/spring-projects/spring-amqp/issues/432/comments","events_url":"https://api.github.com/repos/spring-projects/spring-amqp/issues/432/events","html_url":"https://github.com/spring-projects/spring-amqp/pull/432","id":165226167,"number":432,"title":"AMQP-621: Fix Consumer.basicCancel race condition","user":{"login":"artembilan","id":1071590,"avatar_url":"https://avatars.githubusercontent.com/u/1071590?v=3","gravatar_id":"","url":"https://api.github.com/users/artembilan","html_url":"https://github.com/artembilan","followers_url":"https://api.github.com/users/artembilan/followers","following_url":"https://api.github.com/users/artembilan/following{/other_user}","gists_url":"https://api.github.com/users/artembilan/gists{/gist_id}","starred_url":"https://api.github.com/users/artembilan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/artembilan/subscriptions","organizations_url":"https://api.github.com/users/artembilan/orgs","repos_url":"https://api.github.com/users/artembilan/repos","events_url":"https://api.github.com/users/artembilan/events{/privacy}","received_events_url":"https://api.github.com/users/artembilan/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2016-07-13T02:59:55Z","updated_at":"2016-07-13T08:58:50Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/spring-projects/spring-amqp/pulls/432","html_url":"https://github.com/spring-projects/spring-amqp/pull/432","diff_url":"https://github.com/spring-projects/spring-amqp/pull/432.diff","patch_url":"https://github.com/spring-projects/spring-amqp/pull/432.patch"},"body":"JIRA: https://jira.spring.io/browse/AMQP-621\r\n\r\nThe `SimpleMessageListenerContainer` has logic to initiate `BlockingQueueConsumer.basicCancel()` and mark `BlockingQueueConsumer` as inactive immediately.\r\nBut the `handleCancelOk` answer can come back from the Broker a bit late, in async manner.\r\nDuring this time window we can still receive messages via `handleDelivery`.\r\nWith `prefetchCount = 1` we may end up with the deadlock on `handleDelivery` thread during that time window, when main consumer loop identifies consumer as inactive and exits for `stop` process\r\nand can't do that because the `BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body))` operation blocks the consumer from any other operations.\r\n\r\n* Fix the race condition via additional `BlockingQueueConsumer.cancelled()` state to let main loop to spin until `handleCancelOk` honoring any in-flight messages\r\n* The `cancelled` state is based on the `cancelReceived` flag which is changed from the `handleCancelOk` as a Broker response for the `basicCancel`\r\nand in the `handleCancel` event from the Broker as a reason of some unexpected state on Broker, e.g. consuming queue has been removed\r\n* Protect message lost via `basicReject` in case of late arrival into `handleDelivery`, when the consumer has been cancelled already"},"comment":{"url":"https://api.github.com/repos/spring-projects/spring-amqp/issues/comments/232297879","html_url":"https://github.com/spring-projects/spring-amqp/pull/432#issuecomment-232297879","issue_url":"https://api.github.com/repos/spring-projects/spring-amqp/issues/432","id":232297879,"user":{"login":"garyrussell","id":483832,"avatar_url":"https://avatars.githubusercontent.com/u/483832?v=3","gravatar_id":"","url":"https://api.github.com/users/garyrussell","html_url":"https://github.com/garyrussell","followers_url":"https://api.github.com/users/garyrussell/followers","following_url":"https://api.github.com/users/garyrussell/following{/other_user}","gists_url":"https://api.github.com/users/garyrussell/gists{/gist_id}","starred_url":"https://api.github.com/users/garyrussell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/garyrussell/subscriptions","organizations_url":"https://api.github.com/users/garyrussell/orgs","repos_url":"https://api.github.com/users/garyrussell/repos","events_url":"https://api.github.com/users/garyrussell/events{/privacy}","received_events_url":"https://api.github.com/users/garyrussell/received_events","type":"User","site_admin":false},"created_at":"2016-07-13T08:58:50Z","updated_at":"2016-07-13T08:58:50Z","body":"I think it's a bit more complicated than this...\r\n\r\nWhen we receive a `basicCancel` for one of the consumer tags, we need to abort the consumer - including cancelling all the other consumer tags (call `basicCancel()` like we do in `queuesChanged()` - in fact, deleting a queue from the broker is quite similar to changing the containers queues manually.\r\n\r\nWhen we issue `basicCancel`s for each consumer in `basicCancel()`  , we must return `false` from `cancelled()` while `this.consumerTags.size() > 0` - i.e. we need to keep the loop running until all the consumer tags have have been successfully cancelled."}}
{"actor":{"display_login":"garyrussell"}}