{"action":"created","comment":{"url":"https://api.github.com/repos/allegro/hermes/pulls/comments/67837794","id":67837794,"diff_hunk":"@@ -0,0 +1,156 @@\n+package pl.allegro.tech.hermes.consumers.supervisor.process;\n+\n+import org.jctools.queues.MpscArrayQueue;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+import pl.allegro.tech.hermes.api.SubscriptionName;\n+import pl.allegro.tech.hermes.common.config.ConfigFactory;\n+import pl.allegro.tech.hermes.common.config.Configs;\n+import pl.allegro.tech.hermes.consumers.consumer.Consumer;\n+import pl.allegro.tech.hermes.consumers.supervisor.ConsumersExecutorService;\n+\n+import java.time.Clock;\n+import java.util.ArrayList;\n+import java.util.List;\n+import java.util.Optional;\n+import java.util.concurrent.Future;\n+\n+public class ConsumerProcessSupervisor implements Runnable {\n+\n+    private static final Logger logger = LoggerFactory.getLogger(ConsumerProcessSupervisor.class);\n+\n+    private final MpscArrayQueue<Signal> taskQueue = new MpscArrayQueue<>(1000);\n+\n+    private final RunningConsumerProcesses runningProcesses = new RunningConsumerProcesses();\n+\n+    private final Retransmitter retransmitter;\n+\n+    private final ConsumersExecutorService executor;\n+\n+    private final Clock clock;\n+\n+    private long unhealthyAfter;\n+\n+    public ConsumerProcessSupervisor(ConsumersExecutorService executor,\n+                                     Retransmitter retransmitter,\n+                                     Clock clock,\n+                                     ConfigFactory configs) {\n+        this.executor = executor;\n+        this.retransmitter = retransmitter;\n+        this.clock = clock;\n+        this.unhealthyAfter = configs.getIntProperty(Configs.CONSUMER_BACKGROUND_SUPERVISOR_UNHEALTHY_AFTER);\n+    }\n+\n+    public void accept(Signal signal) {\n+        taskQueue.offer(signal);\n+    }\n+\n+    @Override\n+    public void run() {\n+        logger.info(\"Starting process supervisor loop\");\n+        long currentTime = clock.millis();\n+\n+        try {\n+            checkHealth();\n+            taskQueue.drain(this::processSignal);\n+        }\n+        catch(Exception exception) {\n+            logger.error(\"Process supervisor loop failed\", exception);\n+        }\n+\n+        logger.info(\"Process supervisor loop took {} ms to check all consumers\", clock.millis() - currentTime);\n+    }\n+\n+    private void checkHealth() {","path":"hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/ConsumerProcessSupervisor.java","position":64,"original_position":64,"commit_id":"5421fbde84872550d3d48647b50ca2408ed7c901","original_commit_id":"5421fbde84872550d3d48647b50ca2408ed7c901","user":{"login":"wojtkiewicz","id":11539876,"avatar_url":"https://avatars.githubusercontent.com/u/11539876?v=3","gravatar_id":"","url":"https://api.github.com/users/wojtkiewicz","html_url":"https://github.com/wojtkiewicz","followers_url":"https://api.github.com/users/wojtkiewicz/followers","following_url":"https://api.github.com/users/wojtkiewicz/following{/other_user}","gists_url":"https://api.github.com/users/wojtkiewicz/gists{/gist_id}","starred_url":"https://api.github.com/users/wojtkiewicz/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wojtkiewicz/subscriptions","organizations_url":"https://api.github.com/users/wojtkiewicz/orgs","repos_url":"https://api.github.com/users/wojtkiewicz/repos","events_url":"https://api.github.com/users/wojtkiewicz/events{/privacy}","received_events_url":"https://api.github.com/users/wojtkiewicz/received_events","type":"User","site_admin":false},"body":"I think that this should be called `restartUnhealthy` because just checking imply that decision what to do is postponed after processing result. ","created_at":"2016-06-21T09:40:04Z","updated_at":"2016-06-21T09:40:04Z","html_url":"https://github.com/allegro/hermes/pull/505#discussion_r67837794","pull_request_url":"https://api.github.com/repos/allegro/hermes/pulls/505","_links":{"self":{"href":"https://api.github.com/repos/allegro/hermes/pulls/comments/67837794"},"html":{"href":"https://github.com/allegro/hermes/pull/505#discussion_r67837794"},"pull_request":{"href":"https://api.github.com/repos/allegro/hermes/pulls/505"}}},"pull_request":{"url":"https://api.github.com/repos/allegro/hermes/pulls/505","id":73728435,"html_url":"https://github.com/allegro/hermes/pull/505","diff_url":"https://github.com/allegro/hermes/pull/505.diff","patch_url":"https://github.com/allegro/hermes/pull/505.patch","issue_url":"https://api.github.com/repos/allegro/hermes/issues/505","number":505,"state":"open","locked":false,"title":"467 assignment executor blocked","user":{"login":"adamdubiel","id":5269406,"avatar_url":"https://avatars.githubusercontent.com/u/5269406?v=3","gravatar_id":"","url":"https://api.github.com/users/adamdubiel","html_url":"https://github.com/adamdubiel","followers_url":"https://api.github.com/users/adamdubiel/followers","following_url":"https://api.github.com/users/adamdubiel/following{/other_user}","gists_url":"https://api.github.com/users/adamdubiel/gists{/gist_id}","starred_url":"https://api.github.com/users/adamdubiel/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/adamdubiel/subscriptions","organizations_url":"https://api.github.com/users/adamdubiel/orgs","repos_url":"https://api.github.com/users/adamdubiel/repos","events_url":"https://api.github.com/users/adamdubiel/events{/privacy}","received_events_url":"https://api.github.com/users/adamdubiel/received_events","type":"User","site_admin":false},"body":"Revamped version of consumers - less locks & blocks, more queues and signals.\r\n\r\nIn general it is the idea of @wojtkiewicz - making Consumers more like supervised processes with signal-based communication. The heart of whole design are `NonblockingConsumersSupervisor`, `ConsumerProcessSupervisor` and `ConsumerProcess`.\r\n\r\n`ConsumerProcessSupervisor` manages all `ConsumerProcess` processes. Each `ConsumerProcess` is single thread running `Consumer` and all changes (like stop/start/retransmit) take place in this thread. This makes `Consumer` single threaded (in simple cases). Likewise, `ConsumerProcessSupervisor` is for the most part single threaded, except from signal queue. `ConsumerProcessSupervisor` also does some simple health checking. Health check is a simple timestamp regenerated each time signals are processed by `ConsumerProcess`. If `ConsumerProcess` is unable to respond to signals, we assume it got blocked and it gets restarted.\r\n\r\nThose changes forced two other: offset commits and Zookeeper caches.\r\n\r\nOffset commits are no longer pulled from Consumers, as it was before. When Consumer reads the message, it pushes information about inflight message to inflight offsets queue in `OffsetQueue` class. Once it is ready to commit, the info is pushed to ready-to-commit queue in `OffsetQueue`. The algorithm which consumes this data and does the commits is located in `OffsetCommitter` class.\r\n\r\nLast change are Zookeeper caches. Previous implementation made it impossible to listen to both Subscription and Topic events. New `HierarchicalCache` structure makes it easy. Also new abstraction - `InternalNotificationBus` was introduced, which can be used instead in multiple places instead of direct cache traversal.\r\n\r\n*I will do squash & merge via github button when it is ready to merge*","created_at":"2016-06-14T12:12:19Z","updated_at":"2016-06-21T09:40:04Z","closed_at":null,"merged_at":null,"merge_commit_sha":"394ce878a872a60e58aa2eaf12809a92ca21de5c","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/allegro/hermes/pulls/505/commits","review_comments_url":"https://api.github.com/repos/allegro/hermes/pulls/505/comments","review_comment_url":"https://api.github.com/repos/allegro/hermes/pulls/comments{/number}","comments_url":"https://api.github.com/repos/allegro/hermes/issues/505/comments","statuses_url":"https://api.github.com/repos/allegro/hermes/statuses/5421fbde84872550d3d48647b50ca2408ed7c901","head":{"label":"allegro:467_assignment_executor_blocked","ref":"467_assignment_executor_blocked","sha":"5421fbde84872550d3d48647b50ca2408ed7c901","user":{"login":"allegro","id":562236,"avatar_url":"https://avatars.githubusercontent.com/u/562236?v=3","gravatar_id":"","url":"https://api.github.com/users/allegro","html_url":"https://github.com/allegro","followers_url":"https://api.github.com/users/allegro/followers","following_url":"https://api.github.com/users/allegro/following{/other_user}","gists_url":"https://api.github.com/users/allegro/gists{/gist_id}","starred_url":"https://api.github.com/users/allegro/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allegro/subscriptions","organizations_url":"https://api.github.com/users/allegro/orgs","repos_url":"https://api.github.com/users/allegro/repos","events_url":"https://api.github.com/users/allegro/events{/privacy}","received_events_url":"https://api.github.com/users/allegro/received_events","type":"Organization","site_admin":false},"repo":{"id":35662043,"name":"hermes","full_name":"allegro/hermes","owner":{"login":"allegro","id":562236,"avatar_url":"https://avatars.githubusercontent.com/u/562236?v=3","gravatar_id":"","url":"https://api.github.com/users/allegro","html_url":"https://github.com/allegro","followers_url":"https://api.github.com/users/allegro/followers","following_url":"https://api.github.com/users/allegro/following{/other_user}","gists_url":"https://api.github.com/users/allegro/gists{/gist_id}","starred_url":"https://api.github.com/users/allegro/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allegro/subscriptions","organizations_url":"https://api.github.com/users/allegro/orgs","repos_url":"https://api.github.com/users/allegro/repos","events_url":"https://api.github.com/users/allegro/events{/privacy}","received_events_url":"https://api.github.com/users/allegro/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/allegro/hermes","description":"Fast and reliable message broker built on top of Kafka.","fork":false,"url":"https://api.github.com/repos/allegro/hermes","forks_url":"https://api.github.com/repos/allegro/hermes/forks","keys_url":"https://api.github.com/repos/allegro/hermes/keys{/key_id}","collaborators_url":"https://api.github.com/repos/allegro/hermes/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/allegro/hermes/teams","hooks_url":"https://api.github.com/repos/allegro/hermes/hooks","issue_events_url":"https://api.github.com/repos/allegro/hermes/issues/events{/number}","events_url":"https://api.github.com/repos/allegro/hermes/events","assignees_url":"https://api.github.com/repos/allegro/hermes/assignees{/user}","branches_url":"https://api.github.com/repos/allegro/hermes/branches{/branch}","tags_url":"https://api.github.com/repos/allegro/hermes/tags","blobs_url":"https://api.github.com/repos/allegro/hermes/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/allegro/hermes/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/allegro/hermes/git/refs{/sha}","trees_url":"https://api.github.com/repos/allegro/hermes/git/trees{/sha}","statuses_url":"https://api.github.com/repos/allegro/hermes/statuses/{sha}","languages_url":"https://api.github.com/repos/allegro/hermes/languages","stargazers_url":"https://api.github.com/repos/allegro/hermes/stargazers","contributors_url":"https://api.github.com/repos/allegro/hermes/contributors","subscribers_url":"https://api.github.com/repos/allegro/hermes/subscribers","subscription_url":"https://api.github.com/repos/allegro/hermes/subscription","commits_url":"https://api.github.com/repos/allegro/hermes/commits{/sha}","git_commits_url":"https://api.github.com/repos/allegro/hermes/git/commits{/sha}","comments_url":"https://api.github.com/repos/allegro/hermes/comments{/number}","issue_comment_url":"https://api.github.com/repos/allegro/hermes/issues/comments{/number}","contents_url":"https://api.github.com/repos/allegro/hermes/contents/{+path}","compare_url":"https://api.github.com/repos/allegro/hermes/compare/{base}...{head}","merges_url":"https://api.github.com/repos/allegro/hermes/merges","archive_url":"https://api.github.com/repos/allegro/hermes/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/allegro/hermes/downloads","issues_url":"https://api.github.com/repos/allegro/hermes/issues{/number}","pulls_url":"https://api.github.com/repos/allegro/hermes/pulls{/number}","milestones_url":"https://api.github.com/repos/allegro/hermes/milestones{/number}","notifications_url":"https://api.github.com/repos/allegro/hermes/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/allegro/hermes/labels{/name}","releases_url":"https://api.github.com/repos/allegro/hermes/releases{/id}","deployments_url":"https://api.github.com/repos/allegro/hermes/deployments","created_at":"2015-05-15T08:11:16Z","updated_at":"2016-06-09T12:44:12Z","pushed_at":"2016-06-20T07:46:29Z","git_url":"git://github.com/allegro/hermes.git","ssh_url":"git@github.com:allegro/hermes.git","clone_url":"https://github.com/allegro/hermes.git","svn_url":"https://github.com/allegro/hermes","homepage":"http://hermes.allegro.tech","size":4960,"stargazers_count":183,"watchers_count":183,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":35,"mirror_url":null,"open_issues_count":36,"forks":35,"open_issues":36,"watchers":183,"default_branch":"master"}},"base":{"label":"allegro:develop","ref":"develop","sha":"bb99fa05eb97a456c19da53667168a775ebb8337","user":{"login":"allegro","id":562236,"avatar_url":"https://avatars.githubusercontent.com/u/562236?v=3","gravatar_id":"","url":"https://api.github.com/users/allegro","html_url":"https://github.com/allegro","followers_url":"https://api.github.com/users/allegro/followers","following_url":"https://api.github.com/users/allegro/following{/other_user}","gists_url":"https://api.github.com/users/allegro/gists{/gist_id}","starred_url":"https://api.github.com/users/allegro/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allegro/subscriptions","organizations_url":"https://api.github.com/users/allegro/orgs","repos_url":"https://api.github.com/users/allegro/repos","events_url":"https://api.github.com/users/allegro/events{/privacy}","received_events_url":"https://api.github.com/users/allegro/received_events","type":"Organization","site_admin":false},"repo":{"id":35662043,"name":"hermes","full_name":"allegro/hermes","owner":{"login":"allegro","id":562236,"avatar_url":"https://avatars.githubusercontent.com/u/562236?v=3","gravatar_id":"","url":"https://api.github.com/users/allegro","html_url":"https://github.com/allegro","followers_url":"https://api.github.com/users/allegro/followers","following_url":"https://api.github.com/users/allegro/following{/other_user}","gists_url":"https://api.github.com/users/allegro/gists{/gist_id}","starred_url":"https://api.github.com/users/allegro/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allegro/subscriptions","organizations_url":"https://api.github.com/users/allegro/orgs","repos_url":"https://api.github.com/users/allegro/repos","events_url":"https://api.github.com/users/allegro/events{/privacy}","received_events_url":"https://api.github.com/users/allegro/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/allegro/hermes","description":"Fast and reliable message broker built on top of Kafka.","fork":false,"url":"https://api.github.com/repos/allegro/hermes","forks_url":"https://api.github.com/repos/allegro/hermes/forks","keys_url":"https://api.github.com/repos/allegro/hermes/keys{/key_id}","collaborators_url":"https://api.github.com/repos/allegro/hermes/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/allegro/hermes/teams","hooks_url":"https://api.github.com/repos/allegro/hermes/hooks","issue_events_url":"https://api.github.com/repos/allegro/hermes/issues/events{/number}","events_url":"https://api.github.com/repos/allegro/hermes/events","assignees_url":"https://api.github.com/repos/allegro/hermes/assignees{/user}","branches_url":"https://api.github.com/repos/allegro/hermes/branches{/branch}","tags_url":"https://api.github.com/repos/allegro/hermes/tags","blobs_url":"https://api.github.com/repos/allegro/hermes/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/allegro/hermes/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/allegro/hermes/git/refs{/sha}","trees_url":"https://api.github.com/repos/allegro/hermes/git/trees{/sha}","statuses_url":"https://api.github.com/repos/allegro/hermes/statuses/{sha}","languages_url":"https://api.github.com/repos/allegro/hermes/languages","stargazers_url":"https://api.github.com/repos/allegro/hermes/stargazers","contributors_url":"https://api.github.com/repos/allegro/hermes/contributors","subscribers_url":"https://api.github.com/repos/allegro/hermes/subscribers","subscription_url":"https://api.github.com/repos/allegro/hermes/subscription","commits_url":"https://api.github.com/repos/allegro/hermes/commits{/sha}","git_commits_url":"https://api.github.com/repos/allegro/hermes/git/commits{/sha}","comments_url":"https://api.github.com/repos/allegro/hermes/comments{/number}","issue_comment_url":"https://api.github.com/repos/allegro/hermes/issues/comments{/number}","contents_url":"https://api.github.com/repos/allegro/hermes/contents/{+path}","compare_url":"https://api.github.com/repos/allegro/hermes/compare/{base}...{head}","merges_url":"https://api.github.com/repos/allegro/hermes/merges","archive_url":"https://api.github.com/repos/allegro/hermes/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/allegro/hermes/downloads","issues_url":"https://api.github.com/repos/allegro/hermes/issues{/number}","pulls_url":"https://api.github.com/repos/allegro/hermes/pulls{/number}","milestones_url":"https://api.github.com/repos/allegro/hermes/milestones{/number}","notifications_url":"https://api.github.com/repos/allegro/hermes/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/allegro/hermes/labels{/name}","releases_url":"https://api.github.com/repos/allegro/hermes/releases{/id}","deployments_url":"https://api.github.com/repos/allegro/hermes/deployments","created_at":"2015-05-15T08:11:16Z","updated_at":"2016-06-09T12:44:12Z","pushed_at":"2016-06-20T07:46:29Z","git_url":"git://github.com/allegro/hermes.git","ssh_url":"git@github.com:allegro/hermes.git","clone_url":"https://github.com/allegro/hermes.git","svn_url":"https://github.com/allegro/hermes","homepage":"http://hermes.allegro.tech","size":4960,"stargazers_count":183,"watchers_count":183,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":35,"mirror_url":null,"open_issues_count":36,"forks":35,"open_issues":36,"watchers":183,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/allegro/hermes/pulls/505"},"html":{"href":"https://github.com/allegro/hermes/pull/505"},"issue":{"href":"https://api.github.com/repos/allegro/hermes/issues/505"},"comments":{"href":"https://api.github.com/repos/allegro/hermes/issues/505/comments"},"review_comments":{"href":"https://api.github.com/repos/allegro/hermes/pulls/505/comments"},"review_comment":{"href":"https://api.github.com/repos/allegro/hermes/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/allegro/hermes/pulls/505/commits"},"statuses":{"href":"https://api.github.com/repos/allegro/hermes/statuses/5421fbde84872550d3d48647b50ca2408ed7c901"}}}}
{"actor":{"display_login":"wojtkiewicz"}}