@DataRehashed
   public void dataRehashed(DataRehashedEvent<K, V> event) {
      ConsistentHash startHash = event.getConsistentHashAtStart();
      ConsistentHash endHash = event.getConsistentHashAtEnd();
      boolean trace = log.isTraceEnabled();
      if (startHash != null && endHash != null) {
         if (trace) {
            log.tracef("Data rehash occurred startHash: %s and endHash: %s with new topology %s and was pre %s", startHash, endHash,
                    event.getNewTopologyId(), event.isPre());
         }

         if (!changeListener.isEmpty()) {
            if (trace) {
               log.tracef("Previous segments %s ", startHash.getSegmentsForOwner(localAddress));
               log.tracef("After segments %s ", endHash.getSegmentsForOwner(localAddress));
            }
            // we don't care about newly added segments, since that means our run wouldn't include them anyways
            IntSet beforeSegments = IntSets.mutableFrom(startHash.getSegmentsForOwner(localAddress));
            // Now any that were there before but aren't there now should be added - we don't care about new segments
            // since our current request shouldn't be working on it - it will have to retrieve it later
            beforeSegments.removeAll(endHash.getSegmentsForOwner(localAddress));
            if (!beforeSegments.isEmpty()) {
               // We have to make sure all current listeners get the newest hashes updated.  This has to occur for
               // new nodes and nodes leaving as the hash segments will change in both cases.
               for (Map.Entry<Object, SegmentListener> entry : changeListener.entrySet()) {
                  if (trace) {
                     log.tracef("Notifying %s through SegmentChangeListener", entry.getKey());
                  }
                  entry.getValue().lostSegments(beforeSegments);
               }
            } else if (trace) {
               log.tracef("No segments have been removed from data rehash, no notification required");
            }
         } else if (trace) {
            log.tracef("No change listeners present!");
         }
      }
   }
@ DataRehashed public void dataRehashed ( DataRehashedEvent < K , V > event ) { ConsistentHash startHash = event . getConsistentHashAtStart ( ) ; ConsistentHash endHash = event . getConsistentHashAtEnd ( ) ; boolean trace = log . isTraceEnabled ( ) ; if ( startHash != null && endHash != null ) { if ( trace ) { log . tracef ( "Data rehash occurred startHash: %s and endHash: %s with new topology %s and was pre %s" , startHash , endHash , event . getNewTopologyId ( ) , event . isPre ( ) ) ; } if ( ! changeListener . isEmpty ( ) ) { if ( trace ) { log . tracef ( "Previous segments %s " , startHash . getSegmentsForOwner ( localAddress ) ) ; log . tracef ( "After segments %s " , endHash . getSegmentsForOwner ( localAddress ) ) ; } IntSet beforeSegments = IntSets . mutableFrom ( startHash . getSegmentsForOwner ( localAddress ) ) ; beforeSegments . removeAll ( endHash . getSegmentsForOwner ( localAddress ) ) ; if ( ! beforeSegments . isEmpty ( ) ) { for ( Map . Entry < Object , SegmentListener > entry : changeListener . entrySet ( ) ) { if ( trace ) { log . tracef ( "Notifying %s through SegmentChangeListener" , entry . getKey ( ) ) ; } entry . getValue ( ) . lostSegments ( beforeSegments ) ; } } else if ( trace ) { log . tracef ( "No segments have been removed from data rehash, no notification required" ) ; } } else if ( trace ) { log . tracef ( "No change listeners present!" ) ; } } }
We need to listen to data rehash events in case if data moves while we are iterating over it.
If a rehash occurs causing this node to lose a segment and there is something iterating over the stream looking
for values of that segment, we can't guarantee that the data has all been seen correctly, so we must therefore
suspect that node by sending it back to the owner.
@param event The data rehash event