private CompletableFuture < Void > replicateAndInvalidate ( CacheTopology cacheTopology ) { Address nextMember = getNextMember ( cacheTopology ) ; if ( nextMember != null ) { HashSet < Address > otherMembers = new HashSet <> ( cacheTopology . getActualMembers ( ) ) ; Address localAddress = rpcManager . getAddress ( ) ; otherMembers . remove ( localAddress ) ; otherMembers . remove ( nextMember ) ; IntSet oldSegments ; if ( cacheTopology . getCurrentCH ( ) . getMembers ( ) . contains ( localAddress ) ) { oldSegments = IntSets . from ( cacheTopology . getCurrentCH ( ) . getSegmentsForOwner ( localAddress ) ) ; oldSegments . retainAll ( cacheTopology . getPendingCH ( ) . getSegmentsForOwner ( localAddress ) ) ; } else { log . trace ( "Local address is not a member of currentCH, returning" ) ; return CompletableFutures . completedNull ( ) ; } log . trace ( "Segments to replicate and invalidate: " + oldSegments ) ; if ( oldSegments . isEmpty ( ) ) { return CompletableFutures . completedNull ( ) ; } // we'll start at 1, so the counter will never drop to 0 until we send all chunks AtomicInteger outboundInvalidations = new AtomicInteger ( 1 ) ; CompletableFuture < Void > outboundTaskFuture = new CompletableFuture <> ( ) ; OutboundTransferTask outboundTransferTask = new OutboundTransferTask ( nextMember , oldSegments , cacheTopology . getCurrentCH ( ) . getNumSegments ( ) , chunkSize , cacheTopology . getTopologyId ( ) , keyPartitioner , task -> { if ( outboundInvalidations . decrementAndGet ( ) == 0 ) { outboundTaskFuture . complete ( null ) ; } } , chunks -> invalidateChunks ( chunks , otherMembers , outboundInvalidations , outboundTaskFuture , cacheTopology ) , OutboundTransferTask :: defaultMapEntryFromDataContainer , OutboundTransferTask :: defaultMapEntryFromStore , dataContainer , persistenceManager , rpcManager , commandsFactory , entryFactory , timeout , cacheName , true , true ) ; outboundTransferTask . execute ( executorService ) ; return outboundTaskFuture ; } else { return CompletableFutures . completedNull ( ) ; } }
that this node keeps from previous topology .