private byte [ ] generateDeterministicHash ( StreamNode node , Hasher hasher , Map < Integer , byte [ ] > hashes , boolean isChainingEnabled , StreamGraph streamGraph ) { // Include stream node to hash. We use the current size of the computed // hashes as the ID. We cannot use the node's ID, because it is // assigned from a static counter. This will result in two identical // programs having different hashes. generateNodeLocalHash ( hasher , hashes . size ( ) ) ; // Include chained nodes to hash for ( StreamEdge outEdge : node . getOutEdges ( ) ) { if ( isChainable ( outEdge , isChainingEnabled , streamGraph ) ) { // Use the hash size again, because the nodes are chained to // this node. This does not add a hash for the chained nodes. generateNodeLocalHash ( hasher , hashes . size ( ) ) ; } } byte [ ] hash = hasher . hash ( ) . asBytes ( ) ; // Make sure that all input nodes have their hash set before entering // this loop (calling this method). for ( StreamEdge inEdge : node . getInEdges ( ) ) { byte [ ] otherHash = hashes . get ( inEdge . getSourceId ( ) ) ; // Sanity check if ( otherHash == null ) { throw new IllegalStateException ( "Missing hash for input node " + streamGraph . getSourceVertex ( inEdge ) + ". Cannot generate hash for " + node + "." ) ; } for ( int j = 0 ; j < hash . length ; j ++ ) { hash [ j ] = ( byte ) ( hash [ j ] * 37 ^ otherHash [ j ] ) ; } } if ( LOG . isDebugEnabled ( ) ) { String udfClassName = "" ; if ( node . getOperator ( ) instanceof AbstractUdfStreamOperator ) { udfClassName = ( ( AbstractUdfStreamOperator < ? , ? > ) node . getOperator ( ) ) . getUserFunction ( ) . getClass ( ) . getName ( ) ; } LOG . debug ( "Generated hash '" + byteToHexString ( hash ) + "' for node " + "'" + node . toString ( ) + "' {id: " + node . getId ( ) + ", " + "parallelism: " + node . getParallelism ( ) + ", " + "user function: " + udfClassName + "}" ) ; } return hash ; }
Generates a deterministic hash from node - local properties and input and output edges .