[GitHub] [samza] cameronlee314 commented on a change in pull request #1042: SAMZA-2201 : Startpoints - Integrate fan out with job coordinators

[GitBox <gi...@apache.org>]

cameronlee314 commented on a change in pull request #1042: SAMZA-2201 : Startpoints - Integrate fan out with job coordinators
URL: https://github.com/apache/samza/pull/1042#discussion_r286150609
 
 

 ##########
 File path: samza-core/src/main/java/org/apache/samza/job/model/JobModel.java
 ##########
 @@ -119,6 +125,30 @@ public String getContainerToHostValue(String containerId, String key) {
     return allValues;
   }
 
+  /**
+   * Extracts the map of {@link SystemStreamPartition}s to {@link TaskName} from the underlying {@link TaskModel}s
+   *
+   * @return the extracted map
+   */
+  public Map<TaskName, Set<SystemStreamPartition>> getTaskToSystemStreamPartitions() {
+    HashMap<TaskName, Set<SystemStreamPartition>> taskToSSPs = new HashMap<>();
+    for (String containerID : containers.keySet()) {
+      ContainerModel containerModel = containers.get(containerID);
+      for (TaskName taskName : containerModel.getTasks().keySet()) {
+        TaskModel taskModel = containerModel.getTasks().get(taskName);
+        if (taskModel.getTaskMode() != TaskMode.Active) {
+          // Avoid duplicate tasks
+          continue;
+        }
+        ImmutableSet.Builder<SystemStreamPartition> setOfSSPs = ImmutableSet.builder();
+        setOfSSPs.addAll(taskToSSPs.getOrDefault(taskName, ImmutableSet.of()));
 
 Review comment:
   What is the use case for this line? Is `taskName` not unique across containers?

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services