----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/47247/#review132737 -----------------------------------------------------------
It seems that all we needed to fix the issue is to remove the line to register the coordinator stream consumer? Did I miss any other things? samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskAssignmentManager.java (line 53) <https://reviews.apache.org/r/47247/#comment196985> nit: Is this to avoid the call to register() with null input argument? It does not seem to be absolutely needed and makes this CoordinatorStreamManager's lifecycle different from others. samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskAssignmentManager.java (line 58) <https://reviews.apache.org/r/47247/#comment196984> There is no place that we change the registered flag. Why do we need to have it if registered is always false? - Yi Pan (Data Infrastructure) On May 11, 2016, 6:19 p.m., Jake Maes wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/47247/ > ----------------------------------------------------------- > > (Updated May 11, 2016, 6:19 p.m.) > > > Review request for samza, Boris Shkolnik, Chris Pettitt, Jake Maes, Navina > Ramesh, Jagadish Venkatraman, Xinyu Liu, and Yi Pan (Data Infrastructure). > > > Bugs: SAMZA-947 > https://issues.apache.org/jira/browse/SAMZA-947 > > > Repository: samza > > > Description > ------- > > SAMZA-947 - TaskAssignmentManager registration exception when partition count > changes. > > * Register the producer once at constructor time > * Don't bother registering the consumer. It's lifecycle is outside the > TaskAssignmentManager and registering from the TAM ends up being a no-op. > > > Diffs > ----- > > > samza-core/src/main/java/org/apache/samza/container/grouper/task/GroupByContainerCount.java > 286ea1b3d0d39ebd7d9923a81c02c1d0842b1291 > > samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskAssignmentManager.java > 0cbdec8ac050de18c2fea191e3ef38273f1dbab1 > > samza-core/src/test/java/org/apache/samza/container/grouper/task/TestTaskAssignmentManager.java > 19ab78e891ca22b6fba430ded6b9382c860a212d > > Diff: https://reviews.apache.org/r/47247/diff/ > > > Testing > ------- > > Manually tested with a job. Adjusted the input topics to induce a partition > change and verified no exceptions and the TaskAssignmentManager cleanup ran > appropriately. > > 2016-05-11 17:34:33 GroupByContainerCount [WARN] Current task count 32 does > not match saved task count 512. Stateful jobs may observe misalignment of > keys! > ... > 2016-05-11 17:34:34 TaskAssignmentManager [INFO] Task "Partition 15" moved > from container 57 to container null > 2016-05-11 17:34:34 TaskAssignmentManager [INFO] Task "Partition 14" moved > from container 46 to container null > 2016-05-11 17:34:34 TaskAssignmentManager [INFO] Task "Partition 13" moved > from container 35 to container null > 2016-05-11 17:34:34 TaskAssignmentManager [INFO] Task "Partition 19" moved > from container 101 to container null > ... > > > Thanks, > > Jake Maes > >
