private boolean generateNodeHash ( StreamNode node , HashFunction hashFunction , Map < Integer , byte [ ] > hashes , boolean isChainingEnabled , StreamGraph streamGraph ) { // Check for user-specified ID String userSpecifiedHash = node . getTransformationUID ( ) ; if ( userSpecifiedHash == null ) { // Check that all input nodes have their hashes computed for ( StreamEdge inEdge : node . getInEdges ( ) ) { // If the input node has not been visited yet, the current // node will be visited again at a later point when all input // nodes have been visited and their hashes set. if ( ! hashes . containsKey ( inEdge . getSourceId ( ) ) ) { return false ; } } Hasher hasher = hashFunction . newHasher ( ) ; byte [ ] hash = generateDeterministicHash ( node , hasher , hashes , isChainingEnabled , streamGraph ) ; if ( hashes . put ( node . getId ( ) , hash ) != null ) { // Sanity check throw new IllegalStateException ( "Unexpected state. Tried to add node hash " + "twice. This is probably a bug in the JobGraph generator." ) ; } return true ; } else { Hasher hasher = hashFunction . newHasher ( ) ; byte [ ] hash = generateUserSpecifiedHash ( node , hasher ) ; for ( byte [ ] previousHash : hashes . values ( ) ) { if ( Arrays . equals ( previousHash , hash ) ) { throw new IllegalArgumentException ( "Hash collision on user-specified ID " + "\"" + userSpecifiedHash + "\". " + "Most likely cause is a non-unique ID. Please check that all IDs " + "specified via `uid(String)` are unique." ) ; } } if ( hashes . put ( node . getId ( ) , hash ) != null ) { // Sanity check throw new IllegalStateException ( "Unexpected state. Tried to add node hash " + "twice. This is probably a bug in the JobGraph generator." ) ; } return true ; } }
Generates a hash for the node and returns whether the operation was successful .