{"action":"created","comment":{"url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/comments/180271648","pull_request_review_id":110663939,"id":180271648,"diff_hunk":"@@ -38,7 +74,78 @@ public Nukleus create(\n     {\n         KafkaConfiguration kafkaConfig = new KafkaConfiguration(config);\n \n-        return builder.streamFactory(CLIENT, new ClientStreamFactoryBuilder(kafkaConfig))\n+        Consumer<BiFunction<String, Long, NetworkConnectionPool>> connectionPoolFactoryConsumer =\n+            DEFAULT_CONNECT_POOL_FACTORY_CONSUMER;\n+        MessagePredicate routeHandler = DEFAULT_ROUTE_HANDLER;\n+\n+        if (kafkaConfig.bootstrapEnabled())\n+        {\n+            routeHandler = this::handleRouteForBootstrap;\n+            connectionPoolFactoryConsumer = f -> createNetworkConnectionPool = f;\n+        }\n+\n+        ClientStreamFactoryBuilder streamFactoryBuilder = new ClientStreamFactoryBuilder(kafkaConfig,\n+                connectionPools, connectionPoolFactoryConsumer);\n+\n+        return builder.streamFactory(CLIENT, streamFactoryBuilder)\n+                      .routeHandler(CLIENT, routeHandler)\n+                      .inject(this)\n                       .build();\n     }\n+\n+    public boolean handleRouteForBootstrap(int msgTypeId, DirectBuffer buffer, int index, int length)\n+    {\n+        boolean result = true;\n+        switch(msgTypeId)\n+        {\n+        case RouteFW.TYPE_ID:\n+            {\n+                buffer.getBytes(index, pendingRouteBuffer, 0, length);\n+                pendingRoute = routeRO.wrap(pendingRouteBuffer, 0, length);\n+            }\n+            break;\n+        default:\n+            break;\n+        }\n+        return result;\n+    }\n+\n+    @Override\n+    public int process()\n+    {\n+        if (pendingRoute != null && createNetworkConnectionPool != null)\n+        {\n+            final RouteFW route = pendingRoute;\n+            pendingRoute = null;\n+            startTopicBootstrap(route);\n+        }\n+        return 0;\n+    }\n+\n+    public void startTopicBootstrap(RouteFW route)","path":"src/main/java/org/reaktivity/nukleus/kafka/internal/KafkaNukleusFactorySpi.java","position":103,"original_position":107,"commit_id":"db99aae54c5f2c1c42182d40991d298d2a9bbb17","original_commit_id":"722089ff0b1f343aa70a6183e31a4cf3d4ed19d3","user":{"login":"cmebarrow","id":3606936,"avatar_url":"https://avatars3.githubusercontent.com/u/3606936?v=4","gravatar_id":"","url":"https://api.github.com/users/cmebarrow","html_url":"https://github.com/cmebarrow","followers_url":"https://api.github.com/users/cmebarrow/followers","following_url":"https://api.github.com/users/cmebarrow/following{/other_user}","gists_url":"https://api.github.com/users/cmebarrow/gists{/gist_id}","starred_url":"https://api.github.com/users/cmebarrow/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cmebarrow/subscriptions","organizations_url":"https://api.github.com/users/cmebarrow/orgs","repos_url":"https://api.github.com/users/cmebarrow/repos","events_url":"https://api.github.com/users/cmebarrow/events{/privacy}","received_events_url":"https://api.github.com/users/cmebarrow/received_events","type":"User","site_admin":false},"body":"done","created_at":"2018-04-10T00:58:37Z","updated_at":"2018-04-10T00:58:37Z","html_url":"https://github.com/reaktivity/nukleus-kafka.java/pull/73#discussion_r180271648","pull_request_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/comments/180271648"},"html":{"href":"https://github.com/reaktivity/nukleus-kafka.java/pull/73#discussion_r180271648"},"pull_request":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73"}},"in_reply_to_id":180259662},"pull_request":{"url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73","id":180442976,"html_url":"https://github.com/reaktivity/nukleus-kafka.java/pull/73","diff_url":"https://github.com/reaktivity/nukleus-kafka.java/pull/73.diff","patch_url":"https://github.com/reaktivity/nukleus-kafka.java/pull/73.patch","issue_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues/73","number":73,"state":"open","locked":false,"title":"Added bootstrap for compacted topics","user":{"login":"cmebarrow","id":3606936,"avatar_url":"https://avatars3.githubusercontent.com/u/3606936?v=4","gravatar_id":"","url":"https://api.github.com/users/cmebarrow","html_url":"https://github.com/cmebarrow","followers_url":"https://api.github.com/users/cmebarrow/followers","following_url":"https://api.github.com/users/cmebarrow/following{/other_user}","gists_url":"https://api.github.com/users/cmebarrow/gists{/gist_id}","starred_url":"https://api.github.com/users/cmebarrow/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cmebarrow/subscriptions","organizations_url":"https://api.github.com/users/cmebarrow/orgs","repos_url":"https://api.github.com/users/cmebarrow/repos","events_url":"https://api.github.com/users/cmebarrow/events{/privacy}","received_events_url":"https://api.github.com/users/cmebarrow/received_events","type":"User","site_admin":false},"body":"Topics which have explicit routes are automatically bootstrapped as soon as they are routed, if they are compacted. Bootstrap consists of reading all messages in order to populate the key offsets cache (in CachingKeyMessageDispatcher).\r\n\r\n Requires https://github.com/reaktivity/nukleus-kafka.spec/pull/42","created_at":"2018-04-09T22:19:36Z","updated_at":"2018-04-10T00:58:37Z","closed_at":null,"merged_at":null,"merge_commit_sha":"ae0268659abfdb49412ae366d4162f9a1989eaaa","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73/commits","review_comments_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73/comments","review_comment_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/comments{/number}","comments_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues/73/comments","statuses_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/statuses/db99aae54c5f2c1c42182d40991d298d2a9bbb17","head":{"label":"cmebarrow:bootstrap","ref":"bootstrap","sha":"db99aae54c5f2c1c42182d40991d298d2a9bbb17","user":{"login":"cmebarrow","id":3606936,"avatar_url":"https://avatars3.githubusercontent.com/u/3606936?v=4","gravatar_id":"","url":"https://api.github.com/users/cmebarrow","html_url":"https://github.com/cmebarrow","followers_url":"https://api.github.com/users/cmebarrow/followers","following_url":"https://api.github.com/users/cmebarrow/following{/other_user}","gists_url":"https://api.github.com/users/cmebarrow/gists{/gist_id}","starred_url":"https://api.github.com/users/cmebarrow/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cmebarrow/subscriptions","organizations_url":"https://api.github.com/users/cmebarrow/orgs","repos_url":"https://api.github.com/users/cmebarrow/repos","events_url":"https://api.github.com/users/cmebarrow/events{/privacy}","received_events_url":"https://api.github.com/users/cmebarrow/received_events","type":"User","site_admin":false},"repo":{"id":113369612,"name":"nukleus-kafka.java","full_name":"cmebarrow/nukleus-kafka.java","owner":{"login":"cmebarrow","id":3606936,"avatar_url":"https://avatars3.githubusercontent.com/u/3606936?v=4","gravatar_id":"","url":"https://api.github.com/users/cmebarrow","html_url":"https://github.com/cmebarrow","followers_url":"https://api.github.com/users/cmebarrow/followers","following_url":"https://api.github.com/users/cmebarrow/following{/other_user}","gists_url":"https://api.github.com/users/cmebarrow/gists{/gist_id}","starred_url":"https://api.github.com/users/cmebarrow/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cmebarrow/subscriptions","organizations_url":"https://api.github.com/users/cmebarrow/orgs","repos_url":"https://api.github.com/users/cmebarrow/repos","events_url":"https://api.github.com/users/cmebarrow/events{/privacy}","received_events_url":"https://api.github.com/users/cmebarrow/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/cmebarrow/nukleus-kafka.java","description":"Kafka Nukleus Implementation","fork":true,"url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java","forks_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/forks","keys_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/keys{/key_id}","collaborators_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/teams","hooks_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/hooks","issue_events_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/issues/events{/number}","events_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/events","assignees_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/assignees{/user}","branches_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/branches{/branch}","tags_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/tags","blobs_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/git/refs{/sha}","trees_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/git/trees{/sha}","statuses_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/statuses/{sha}","languages_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/languages","stargazers_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/stargazers","contributors_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/contributors","subscribers_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/subscribers","subscription_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/subscription","commits_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/commits{/sha}","git_commits_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/git/commits{/sha}","comments_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/comments{/number}","issue_comment_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/issues/comments{/number}","contents_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/contents/{+path}","compare_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/compare/{base}...{head}","merges_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/merges","archive_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/downloads","issues_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/issues{/number}","pulls_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/pulls{/number}","milestones_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/milestones{/number}","notifications_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/labels{/name}","releases_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/releases{/id}","deployments_url":"https://api.github.com/repos/cmebarrow/nukleus-kafka.java/deployments","created_at":"2017-12-06T21:33:50Z","updated_at":"2018-04-05T19:05:10Z","pushed_at":"2018-04-10T00:57:23Z","git_url":"git://github.com/cmebarrow/nukleus-kafka.java.git","ssh_url":"git@github.com:cmebarrow/nukleus-kafka.java.git","clone_url":"https://github.com/cmebarrow/nukleus-kafka.java.git","svn_url":"https://github.com/cmebarrow/nukleus-kafka.java","homepage":null,"size":662,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"other","name":"Other","spdx_id":null,"url":null},"forks":0,"open_issues":0,"watchers":0,"default_branch":"develop"}},"base":{"label":"reaktivity:develop","ref":"develop","sha":"40472d4ee86c39d229372f10cd119b6f3422d945","user":{"login":"reaktivity","id":22941769,"avatar_url":"https://avatars0.githubusercontent.com/u/22941769?v=4","gravatar_id":"","url":"https://api.github.com/users/reaktivity","html_url":"https://github.com/reaktivity","followers_url":"https://api.github.com/users/reaktivity/followers","following_url":"https://api.github.com/users/reaktivity/following{/other_user}","gists_url":"https://api.github.com/users/reaktivity/gists{/gist_id}","starred_url":"https://api.github.com/users/reaktivity/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/reaktivity/subscriptions","organizations_url":"https://api.github.com/users/reaktivity/orgs","repos_url":"https://api.github.com/users/reaktivity/repos","events_url":"https://api.github.com/users/reaktivity/events{/privacy}","received_events_url":"https://api.github.com/users/reaktivity/received_events","type":"Organization","site_admin":false},"repo":{"id":105943390,"name":"nukleus-kafka.java","full_name":"reaktivity/nukleus-kafka.java","owner":{"login":"reaktivity","id":22941769,"avatar_url":"https://avatars0.githubusercontent.com/u/22941769?v=4","gravatar_id":"","url":"https://api.github.com/users/reaktivity","html_url":"https://github.com/reaktivity","followers_url":"https://api.github.com/users/reaktivity/followers","following_url":"https://api.github.com/users/reaktivity/following{/other_user}","gists_url":"https://api.github.com/users/reaktivity/gists{/gist_id}","starred_url":"https://api.github.com/users/reaktivity/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/reaktivity/subscriptions","organizations_url":"https://api.github.com/users/reaktivity/orgs","repos_url":"https://api.github.com/users/reaktivity/repos","events_url":"https://api.github.com/users/reaktivity/events{/privacy}","received_events_url":"https://api.github.com/users/reaktivity/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/reaktivity/nukleus-kafka.java","description":"Kafka Nukleus Implementation","fork":false,"url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java","forks_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/forks","keys_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/keys{/key_id}","collaborators_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/teams","hooks_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/hooks","issue_events_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues/events{/number}","events_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/events","assignees_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/assignees{/user}","branches_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/branches{/branch}","tags_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/tags","blobs_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/git/refs{/sha}","trees_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/git/trees{/sha}","statuses_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/statuses/{sha}","languages_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/languages","stargazers_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/stargazers","contributors_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/contributors","subscribers_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/subscribers","subscription_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/subscription","commits_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/commits{/sha}","git_commits_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/git/commits{/sha}","comments_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/comments{/number}","issue_comment_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues/comments{/number}","contents_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/contents/{+path}","compare_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/compare/{base}...{head}","merges_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/merges","archive_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/downloads","issues_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues{/number}","pulls_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls{/number}","milestones_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/milestones{/number}","notifications_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/labels{/name}","releases_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/releases{/id}","deployments_url":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/deployments","created_at":"2017-10-05T21:42:58Z","updated_at":"2018-04-09T19:46:01Z","pushed_at":"2018-04-10T00:57:24Z","git_url":"git://github.com/reaktivity/nukleus-kafka.java.git","ssh_url":"git@github.com:reaktivity/nukleus-kafka.java.git","clone_url":"https://github.com/reaktivity/nukleus-kafka.java.git","svn_url":"https://github.com/reaktivity/nukleus-kafka.java","homepage":null,"size":574,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":true,"has_projects":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":3,"mirror_url":null,"archived":false,"open_issues_count":15,"license":{"key":"other","name":"Other","spdx_id":null,"url":null},"forks":3,"open_issues":15,"watchers":0,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73"},"html":{"href":"https://github.com/reaktivity/nukleus-kafka.java/pull/73"},"issue":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues/73"},"comments":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/issues/73/comments"},"review_comments":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73/comments"},"review_comment":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/pulls/73/commits"},"statuses":{"href":"https://api.github.com/repos/reaktivity/nukleus-kafka.java/statuses/db99aae54c5f2c1c42182d40991d298d2a9bbb17"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"cmebarrow"}}