{"action":"created","issue":{"url":"https://api.github.com/repos/confluentinc/kafka/issues/21","labels_url":"https://api.github.com/repos/confluentinc/kafka/issues/21/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/kafka/issues/21/comments","events_url":"https://api.github.com/repos/confluentinc/kafka/issues/21/events","html_url":"https://github.com/confluentinc/kafka/pull/21","id":103796343,"number":21,"title":"fix context.forward() using a stack of nodes","user":{"login":"ymatsuda","id":169548,"avatar_url":"https://avatars.githubusercontent.com/u/169548?v=3","gravatar_id":"","url":"https://api.github.com/users/ymatsuda","html_url":"https://github.com/ymatsuda","followers_url":"https://api.github.com/users/ymatsuda/followers","following_url":"https://api.github.com/users/ymatsuda/following{/other_user}","gists_url":"https://api.github.com/users/ymatsuda/gists{/gist_id}","starred_url":"https://api.github.com/users/ymatsuda/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ymatsuda/subscriptions","organizations_url":"https://api.github.com/users/ymatsuda/orgs","repos_url":"https://api.github.com/users/ymatsuda/repos","events_url":"https://api.github.com/users/ymatsuda/events{/privacy}","received_events_url":"https://api.github.com/users/ymatsuda/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2015-08-28T21:12:01Z","updated_at":"2015-08-28T22:01:07Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/confluentinc/kafka/pulls/21","html_url":"https://github.com/confluentinc/kafka/pull/21","diff_url":"https://github.com/confluentinc/kafka/pull/21.diff","patch_url":"https://github.com/confluentinc/kafka/pull/21.patch"},"body":"@guozhangwang \r\n\r\nflatMap, flatMapValue and join were broken due to the way context.forward() sets the current node. Any processor that calls forward() multiple times in a single process() call doesn't work since after calling childNode.process() the current node is set to childNode and forget the original caller node. After forward(), the \"current node\" is one of the children, thus subsequent forward() propagates KV to wrong nodes. My solution is use a stack to remember the call hierarchy and set to the correct node after childNode.process(). This is done in ProcessorContextImpl (and MockProcessorContext for testing)\r\n\r\njoinPrior is removed. We can revisit it if necessary.\r\n\r\njoin is now implemented using three processors. Originally I though two are enough. But I realized I need the third one since the processor cannot freely control the recipient of forward().\r\n"},"comment":{"url":"https://api.github.com/repos/confluentinc/kafka/issues/comments/135898166","html_url":"https://github.com/confluentinc/kafka/pull/21#issuecomment-135898166","issue_url":"https://api.github.com/repos/confluentinc/kafka/issues/21","id":135898166,"user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars.githubusercontent.com/u/10729597?v=3","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"created_at":"2015-08-28T22:01:07Z","updated_at":"2015-08-28T22:01:07Z","body":"One minor comment, otherwise LGTM."}}