src/java/org/apache/cassandra/db/transaction/TransactionProxy.java
//WL TODO: Remove the assumption that datacenters have matching keyranges
WL TODO: Remove the assumption that datacenters have matching keyranges
WL TODO : Remove the assumption that datacenters have matching keyranges
private static void replicateTransactionToOtherDatacenters(MessageProducer producer, String keyspace, ByteBuffer targetKey)
    throws IOException
    {
        List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(keyspace, targetKey);
        String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
        IWriteResponseHandler ignoredResponseHandler = WriteResponseHandler.create(endpoints, ConsistencyLevel.ALL, keyspace);
        Map<String, Multimap<Message, InetAddress>> dcMessages = new HashMap<String, Multimap<Message, InetAddress>>(endpoints.size());
        for (InetAddress destination : endpoints) {
            assert FailureDetector.instance.isAlive(destination) : "Not dealing with failed nodes in the dc right now";
            if (destination.equals(FBUtilities.getBroadcastAddress())) {
                continue;
            }
            if (logger.isDebugEnabled())
                logger.debug("replicate transaction to " + destination);
            String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination);
            Multimap<Message, InetAddress> messages = dcMessages.get(dc);
            if (messages == null) {
                messages = HashMultimap.create();
                dcMessages.put(dc, messages);
            }
            messages.put(producer.getMessage(Gossiper.instance.getVersion(destination)), destination);
        }
        StorageProxy.sendMessages(localDataCenter, dcMessages, ignoredResponseHandler);
    }
private static void replicateTransactionToOtherDatacenters ( MessageProducer producer , String keyspace , ByteBuffer targetKey ) throws IOException { List < InetAddress > endpoints = StorageService . instance . getLiveNaturalEndpoints ( keyspace , targetKey ) ; String localDataCenter = DatabaseDescriptor . getEndpointSnitch ( ) . getDatacenter ( FBUtilities . getBroadcastAddress ( ) ) ; IWriteResponseHandler ignoredResponseHandler = WriteResponseHandler . create ( endpoints , ConsistencyLevel . ALL , keyspace ) ; Map < String , Multimap < Message , InetAddress > > dcMessages = new HashMap < String , Multimap < Message , InetAddress > > ( endpoints . size ( ) ) ; for ( InetAddress destination : endpoints ) { assert FailureDetector . instance . isAlive ( destination ) : "Not dealing with failed nodes in the dc right now" ; if ( destination . equals ( FBUtilities . getBroadcastAddress ( ) ) ) { continue ; } if ( logger . isDebugEnabled ( ) ) logger . debug ( "replicate transaction to " + destination ) ; String dc = DatabaseDescriptor . getEndpointSnitch ( ) . getDatacenter ( destination ) ; Multimap < Message , InetAddress > messages = dcMessages . get ( dc ) ; if ( messages == null ) { messages = HashMultimap . create ( ) ; dcMessages . put ( dc , messages ) ; } messages . put ( producer . getMessage ( Gossiper . instance . getVersion ( destination ) ) , destination ) ; } StorageProxy . sendMessages ( localDataCenter , dcMessages , ignoredResponseHandler ) ; }
WL TODO: Remove the assumption that datacenters have matching keyranges
WL TODO : Remove the assumption that datacenters have matching keyranges
// code follows the structure of StorageProxy.sendToHintedEndpoints // Multimap that holds onto all the messages and addresses meant for a specific datacenter //no need to *replicate* to the local machine, it should already have started handling the transaction