GitHub user guozhangwang opened a pull request: https://github.com/apache/kafka/pull/4224
[WIP] KAFKA-6170; KIP-220 Part 2: Break dependency of Assignor on StreamThread This refactoring is discussed in https://github.com/apache/kafka/pull/3624#discussion_r132614639. More specifically: 1. Moved the access of `StreamThread` in `StreamPartitionAssignor` to `TaskManager`, removed any fields stored in `StreamThread` such as `processId` and `clientId` that are only to be used in `StreamPartitionAssignor`, and pass them to `TaskManager` if necessary. 2. Moved any in-memory states, `metadataWithInternalTopics`, `partitionsByHostState`, `standbyTasks`, `activeTasks` to `TaskManager` so that `StreamPartitionAssignor` becomes a stateless thin layer that access TaskManager directly. 3. Remove the reference of `StreamPartitionAssignor` in `StreamThread`, instead consolidate all related functionalities such as `cachedTasksIds ` in `TaskManager` which could be retrieved by the `StreamThread` and the `StreamPartitionAssignor` directly. 4. Finally, removed the two interfaces used for `StreamThread` and `StreamPartitionAssignor`. 5. Some minor fixes on logPrefixes, etc. Future work: when replacing the StreamsKafkaClient, we would let `StreamPartitionAssignor` to retrieve it from `TaskManager` directly, and also its closing call do not need to be called (`KafkaStreams` will be responsible for closing it). ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) You can merge this pull request into a Git repository by running: $ git pull https://github.com/guozhangwang/kafka K6170-refactor-assignor Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/4224.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #4224 ---- commit 6c5b20ea34323a101118286e9282568f428b8e34 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-07T00:14:53Z add AdminClient commit fc908e06d80816db1e28e0f1d05e1d10fa1d0379 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-13T22:13:37Z Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka into K6170-admin-client commit d1be566efe65c71c068a6e948c59f7bd980d6bd8 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-14T21:41:20Z refactor thread / assignor dependency commit d1a778fff0cbaeb8ea00421d89fcd50552b93eba Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-14T21:44:09Z revert TaskManager APIs commit f9e5fbff4c18764bc64793dc9b5c376d956cd67c Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T02:04:55Z move logic of assignor to task manager commit bfd08c45cab067035d4980d85d6e7ff9cd5a6e36 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T02:16:37Z minor fix commit f95dc0bb9849356ab721c4f7e042a813fcb34330 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T02:22:34Z extract delegating restore listener commit 10ceff07c23ea555bd25ea74baa4b995ea0f3a83 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T02:26:59Z add admin configs in streams config commit 41dc2b0790866bb5f8325191102622bdbd5fbe23 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T19:39:29Z add AdminClient to stream thread commit 3592206eb7c06313a7f553242329f6eb578b4cbd Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T22:49:39Z Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka into K6170-admin-client commit 03e64d0bb4a6581d4105f9faa4d95cd6e20f45f3 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-15T23:49:16Z add admin prefix commit 035b3a6a04025d397fec8abb535d9b148f722792 Author: Guozhang Wang <wangg...@gmail.com> Date: 2017-11-16T00:07:19Z merge from K6170-admin-client ---- ---