private Map < Integer , LocalAssignment > getLocalAssign ( StormClusterState stormClusterState , String supervisorId , Map < String , Assignment > assignments ) throws Exception { Map < Integer , LocalAssignment > portToAssignment = new HashMap <> ( ) ; for ( Entry < String , Assignment > assignEntry : assignments . entrySet ( ) ) { String topologyId = assignEntry . getKey ( ) ; Assignment assignment = assignEntry . getValue ( ) ; Map < Integer , LocalAssignment > portTasks = readMyTasks ( stormClusterState , topologyId , supervisorId , assignment ) ; if ( portTasks == null ) { continue ; } // a port must be assigned to one assignment for ( Entry < Integer , LocalAssignment > entry : portTasks . entrySet ( ) ) { Integer port = entry . getKey ( ) ; LocalAssignment la = entry . getValue ( ) ; if ( ! portToAssignment . containsKey ( port ) ) { portToAssignment . put ( port , la ) ; } else { throw new RuntimeException ( "Should not have multiple topologies assigned to one port" ) ; } } } return portToAssignment ; }
a port must be assigned to a topology