{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/221492650","pull_request_review_id":160151606,"id":221492650,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDIyMTQ5MjY1MA==","diff_hunk":"@@ -0,0 +1,158 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements. See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License. You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.kafka.streams.kstream;\n+\n+\n+import org.apache.kafka.common.serialization.Serde;\n+\n+/**\n+ * The class that is used to capture the key and value {@link Serde}s and set the name used for\n+ * repartition topics when performing {@link KStream#groupBy(KeyValueMapper, Grouped)}, {@link\n+ * KStream#groupByKey(Grouped)} {@link KTable#groupBy(KeyValueMapper, Grouped)}operations.\n+ *\n+ * @param <K> the key type\n+ * @param <V> the value type\n+ */\n+public class Grouped<K, V> {\n+\n+    protected Serde<K> keySerde;\n+    protected Serde<V> valueSerde;\n+    protected String name;\n+\n+\n+    private Grouped(final String name,\n+                    final Serde<K> keySerde,\n+                    final Serde<V> valueSerde) {\n+        this.name = name;\n+        this.keySerde = keySerde;\n+        this.valueSerde = valueSerde;\n+    }\n+\n+    protected Grouped(final Grouped<K, V> grouped) {\n+        this(grouped.name, grouped.keySerde, grouped.valueSerde);\n+    }\n+\n+    /**\n+     * Create a Grouped instance with the provided name used for a repartition topic required for\n+     * performing the grouping operation.\n+     *\n+     * @param name the name used for a repartition topic if required\n+     * @return a new {@link Grouped} configured with the name\n+     * @see KStream#groupByKey(Grouped)\n+     * @see KStream#groupBy(KeyValueMapper, Grouped)\n+     * @see KTable#groupBy(KeyValueMapper, Grouped)\n+     */\n+    public static <K, V> Grouped<K, V> named(final String name) {","path":"streams/src/main/java/org/apache/kafka/streams/kstream/Grouped.java","position":60,"original_position":60,"commit_id":"29b93e50b552041b24bbb71af910dfb8fc7d78e0","original_commit_id":"29b93e50b552041b24bbb71af910dfb8fc7d78e0","user":{"login":"bbejeck","id":199238,"node_id":"MDQ6VXNlcjE5OTIzOA==","avatar_url":"https://avatars3.githubusercontent.com/u/199238?v=4","gravatar_id":"","url":"https://api.github.com/users/bbejeck","html_url":"https://github.com/bbejeck","followers_url":"https://api.github.com/users/bbejeck/followers","following_url":"https://api.github.com/users/bbejeck/following{/other_user}","gists_url":"https://api.github.com/users/bbejeck/gists{/gist_id}","starred_url":"https://api.github.com/users/bbejeck/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbejeck/subscriptions","organizations_url":"https://api.github.com/users/bbejeck/orgs","repos_url":"https://api.github.com/users/bbejeck/repos","events_url":"https://api.github.com/users/bbejeck/events{/privacy}","received_events_url":"https://api.github.com/users/bbejeck/received_events","type":"User","site_admin":false},"body":"ack, `Grouped.as` is better, I'll update.","created_at":"2018-10-01T04:14:00Z","updated_at":"2018-10-01T04:14:00Z","html_url":"https://github.com/apache/kafka/pull/5709#discussion_r221492650","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/5709","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/221492650"},"html":{"href":"https://github.com/apache/kafka/pull/5709#discussion_r221492650"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/5709"}},"in_reply_to_id":221470841},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/5709","id":218850040,"node_id":"MDExOlB1bGxSZXF1ZXN0MjE4ODUwMDQw","html_url":"https://github.com/apache/kafka/pull/5709","diff_url":"https://github.com/apache/kafka/pull/5709.diff","patch_url":"https://github.com/apache/kafka/pull/5709.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/5709","number":5709,"state":"open","locked":false,"title":"KAFKA-7406: Name join group repartition topics","user":{"login":"bbejeck","id":199238,"node_id":"MDQ6VXNlcjE5OTIzOA==","avatar_url":"https://avatars3.githubusercontent.com/u/199238?v=4","gravatar_id":"","url":"https://api.github.com/users/bbejeck","html_url":"https://github.com/bbejeck","followers_url":"https://api.github.com/users/bbejeck/followers","following_url":"https://api.github.com/users/bbejeck/following{/other_user}","gists_url":"https://api.github.com/users/bbejeck/gists{/gist_id}","starred_url":"https://api.github.com/users/bbejeck/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbejeck/subscriptions","organizations_url":"https://api.github.com/users/bbejeck/orgs","repos_url":"https://api.github.com/users/bbejeck/repos","events_url":"https://api.github.com/users/bbejeck/events{/privacy}","received_events_url":"https://api.github.com/users/bbejeck/received_events","type":"User","site_admin":false},"body":"This PR has the changes required to implement KIP-372.\r\n\r\n1. Name repartition topics for Joins\r\n2. Name repartition topics for Grouping operations\r\n3. For optimizations where repartition nodes are reduced or merged the topic name of the **_first_** repartition node to be merged is used as the topic name for the new optimized repartition node.\r\n4. Added the `Grouped` class\r\n5. Deprecated the `Serialized` class\r\n6. Updated Joined to accept a `name` parameter for naming repartition topics if required.\r\n6. Updated KStream.groupBy, KStream.groupByKey, KTable.groupBy methods to accept a `Grouped` instance and deprecated methods taking a `Serialized` parameter.\r\n\r\nAlso, our approach when optimizing repartition topic is to **_always_** use the first repartition topic to get merged even if the user has elected to not name any repartition topics.  This approach should make the new topology compatible with existing applications.  However, an application reset may be required as all data may not have been consumed from existing repartition topics.\r\n\r\nI've added tests for the following:\r\n\r\n1. Verify that named repartition topics stay unchanged when topology changes.\r\n2. Verify that user-specified repartition topic names are used over Streams generated names.\r\n2. Verify that when optimization occurs, the new optimized repartition topic is the same name as the **_first_** replaced/merged repartition node.\r\n3. The existing tests validate that when names for repartition topics are not provided the Streams generated names are used. \r\n\r\n### Committer Checklist (excluded from commit message)\r\n- [ ] Verify design and implementation \r\n- [ ] Verify test coverage and CI build status\r\n- [ ] Verify documentation (including upgrade notes)\r\n","created_at":"2018-09-28T04:20:44Z","updated_at":"2018-10-01T04:14:00Z","closed_at":null,"merged_at":null,"merge_commit_sha":"89b675c9d6637edc74932a26138ec5f661e9617d","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/5709/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/5709/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/5709/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/29b93e50b552041b24bbb71af910dfb8fc7d78e0","head":{"label":"bbejeck:KAFKA-7406_name_join_group_repartition_topics","ref":"KAFKA-7406_name_join_group_repartition_topics","sha":"29b93e50b552041b24bbb71af910dfb8fc7d78e0","user":{"login":"bbejeck","id":199238,"node_id":"MDQ6VXNlcjE5OTIzOA==","avatar_url":"https://avatars3.githubusercontent.com/u/199238?v=4","gravatar_id":"","url":"https://api.github.com/users/bbejeck","html_url":"https://github.com/bbejeck","followers_url":"https://api.github.com/users/bbejeck/followers","following_url":"https://api.github.com/users/bbejeck/following{/other_user}","gists_url":"https://api.github.com/users/bbejeck/gists{/gist_id}","starred_url":"https://api.github.com/users/bbejeck/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbejeck/subscriptions","organizations_url":"https://api.github.com/users/bbejeck/orgs","repos_url":"https://api.github.com/users/bbejeck/repos","events_url":"https://api.github.com/users/bbejeck/events{/privacy}","received_events_url":"https://api.github.com/users/bbejeck/received_events","type":"User","site_admin":false},"repo":{"id":42054236,"node_id":"MDEwOlJlcG9zaXRvcnk0MjA1NDIzNg==","name":"kafka","full_name":"bbejeck/kafka","private":false,"owner":{"login":"bbejeck","id":199238,"node_id":"MDQ6VXNlcjE5OTIzOA==","avatar_url":"https://avatars3.githubusercontent.com/u/199238?v=4","gravatar_id":"","url":"https://api.github.com/users/bbejeck","html_url":"https://github.com/bbejeck","followers_url":"https://api.github.com/users/bbejeck/followers","following_url":"https://api.github.com/users/bbejeck/following{/other_user}","gists_url":"https://api.github.com/users/bbejeck/gists{/gist_id}","starred_url":"https://api.github.com/users/bbejeck/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbejeck/subscriptions","organizations_url":"https://api.github.com/users/bbejeck/orgs","repos_url":"https://api.github.com/users/bbejeck/repos","events_url":"https://api.github.com/users/bbejeck/events{/privacy}","received_events_url":"https://api.github.com/users/bbejeck/received_events","type":"User","site_admin":false},"html_url":"https://github.com/bbejeck/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/bbejeck/kafka","forks_url":"https://api.github.com/repos/bbejeck/kafka/forks","keys_url":"https://api.github.com/repos/bbejeck/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/bbejeck/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/bbejeck/kafka/teams","hooks_url":"https://api.github.com/repos/bbejeck/kafka/hooks","issue_events_url":"https://api.github.com/repos/bbejeck/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/bbejeck/kafka/events","assignees_url":"https://api.github.com/repos/bbejeck/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/bbejeck/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/bbejeck/kafka/tags","blobs_url":"https://api.github.com/repos/bbejeck/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/bbejeck/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/bbejeck/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/bbejeck/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/bbejeck/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/bbejeck/kafka/languages","stargazers_url":"https://api.github.com/repos/bbejeck/kafka/stargazers","contributors_url":"https://api.github.com/repos/bbejeck/kafka/contributors","subscribers_url":"https://api.github.com/repos/bbejeck/kafka/subscribers","subscription_url":"https://api.github.com/repos/bbejeck/kafka/subscription","commits_url":"https://api.github.com/repos/bbejeck/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/bbejeck/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/bbejeck/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/bbejeck/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/bbejeck/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/bbejeck/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/bbejeck/kafka/merges","archive_url":"https://api.github.com/repos/bbejeck/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/bbejeck/kafka/downloads","issues_url":"https://api.github.com/repos/bbejeck/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/bbejeck/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/bbejeck/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/bbejeck/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/bbejeck/kafka/labels{/name}","releases_url":"https://api.github.com/repos/bbejeck/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/bbejeck/kafka/deployments","created_at":"2015-09-07T13:23:16Z","updated_at":"2018-09-28T02:02:02Z","pushed_at":"2018-09-28T23:03:14Z","git_url":"git://github.com/bbejeck/kafka.git","ssh_url":"git@github.com:bbejeck/kafka.git","clone_url":"https://github.com/bbejeck/kafka.git","svn_url":"https://github.com/bbejeck/kafka","homepage":null,"size":73545,"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":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"9f7267dd2fedde86bf15aabdbc5256e5fc617184","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"node_id":"MDEwOlJlcG9zaXRvcnkyMjExMjQz","name":"kafka","full_name":"apache/kafka","private":false,"owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2018-10-01T02:53:56Z","pushed_at":"2018-10-01T02:57:18Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":73969,"stargazers_count":9744,"watchers_count":9744,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":5480,"mirror_url":null,"archived":false,"open_issues_count":512,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":5480,"open_issues":512,"watchers":9744,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/5709"},"html":{"href":"https://github.com/apache/kafka/pull/5709"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/5709"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/5709/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/5709/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/5709/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/29b93e50b552041b24bbb71af910dfb8fc7d78e0"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"bbejeck"}}