package org.apache.cassandra.hadoop.trackers;

import java.net.InetAddress;
import java.util.Set;

import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.log4j.Logger;

public class CassandraJobConf extends org.apache.hadoop.mapred.JobConf
{
    private static final Logger logger = Logger.getLogger(CassandraJobConf.class);
    
    public String get(String name, String defaultValue)
    {  
        if (name.equals("mapred.job.tracker") || name.equals("mapreduce.jobtracker.address"))     
        {
            String address = getJobTrackerNode().getHostName()+":8012";
                        
            return address;
        }

        return super.get(name, defaultValue);
    }

    public String get(String name)
    {        
        if (name.equals("mapred.job.tracker") || name.equals("mapreduce.jobtracker.address"))
            return getJobTrackerNode().getHostName()+":8012";
        
        return super.get(name);
    }

    //Will pick a live seed to use as a job tracker in this local dc
    public static InetAddress getJobTrackerNode()
    {
        
        //Get this nodes local DC
        String localDC = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getLocalAddress());
        
        Set<InetAddress> seeds     = DatabaseDescriptor.getSeeds();
        Set<InetAddress> liveNodes = Gossiper.instance.getLiveMembers();

        //Pick a seed in the same DC as this node to be the job tracker
        for (InetAddress seed : seeds)           
            if (liveNodes.contains(seed) && DatabaseDescriptor.getEndpointSnitch().getDatacenter(seed).equals(localDC))               
                return seed;
        
        //Sleep a bit while gossip kicks in
        try
        {
            Thread.sleep(5000);
        }
        catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        }
        
        liveNodes = Gossiper.instance.getLiveMembers();

        for (InetAddress seed : seeds)           
            if (liveNodes.contains(seed) && DatabaseDescriptor.getEndpointSnitch().getDatacenter(seed).equals(localDC))
                return seed;
        
        throw new RuntimeException("No live seeds found in this DC: "+localDC);
    }   
}

@@ -69,16 +69,70 @@
 ddress;%0A
+import java.util.Arrays;%0Aimport java.util.Comparator;%0A
 import j
@@ -1117,21 +1117,16 @@
  pick a 
-live 
 seed to 
@@ -1163,16 +1163,124 @@
 ocal dc%0A
+    //We can't check for live seeds because if this is a ec2 cluster%0A    //the seeds might not be up yet :(%0A
     publ
@@ -1322,26 +1322,24 @@
 Node()%0A    %7B
-%0A 
        %0A    
@@ -1521,17 +1521,16 @@
 eeds    
- 
 = Databa
@@ -1554,36 +1554,41 @@
 eeds();%0A        
-Set%3C
+%0A        
 InetAddress%3E liv
@@ -1586,67 +1586,69 @@
 ress
-%3E liveNodes = Gossiper.instance.getLiveMembers(
+%5B%5D sortedSeeds = seeds.toArray(new InetAddress%5B%5D%7B%7D
 );%0A
-%0A
         
 //Pi
@@ -1647,110 +1647,127 @@
     
-//Pick a seed in the same DC as this node to be the job tracker%0A        for (InetAddress seed : seeds)
+Arrays.sort(sortedSeeds, new Comparator%3CInetAddress%3E()%7B%0A%0A            public int compare(InetAddress a, InetAddress b)%0A 
     
@@ -1765,32 +1765,33 @@
  b)%0A            
+%7B
 %0A            if 
@@ -1791,131 +1791,237 @@
     
-if (liveNodes.contains(seed) && DatabaseDescriptor.getEndpointSnitch().getDatacenter(seed).equals(localDC))               %0A
+    byte a1%5B%5D = a.getAddress();%0A                byte b1%5B%5D = b.getAddress();%0A                %0A                if(a1.length %3C b1.length)%0A                    return -1;%0A                %0A                if(a1.length %3E b1.length)%0A    
     
@@ -2031,36 +2031,33 @@
          return 
-seed
+1
 ;%0A        %0A     
@@ -2042,32 +2042,40 @@
 turn 1;%0A        
+ 
+       
 %0A        //Sleep
@@ -2071,106 +2071,211 @@
     
-//Sleep a bit while gossip kicks in%0A        try%0A        %7B%0A            Thread.sleep(5000);%0A
+        for(int i=0; i%3Ca1.length; i++)%0A                %7B%0A                    if(a1%5Bi%5D %3C b1%5Bi%5D)%0A                        return -1;%0A                    %0A                    if(a1%5Bi%5D %3E b1%5Bi%5D)%0A      
         
 %7D%0A  
@@ -2270,26 +2270,24 @@
         
-%7D%0A
         
 catch (I
@@ -2282,167 +2282,216 @@
     
-catch (InterruptedException e)%0A        %7B%0A            throw new RuntimeException(e);%0A        %7D%0A        %0A        liveNodes = Gossiper.instance.getLiveMembers();%0A
+  return 1;                   %0A                %7D%0A                %0A                return 0;%0A            %7D         %0A        %7D);%0A        %0A%0A        //Pick a seed in the same DC as this node to be the job tracker
 %0A   
@@ -2520,16 +2520,22 @@
 seed : s
+ortedS
 eeds)   
@@ -2563,36 +2563,8 @@
 if (
-liveNodes.contains(seed) && 
 Data
@@ -2634,16 +2634,31 @@
 ocalDC))
+               
 %0A       
@@ -2679,32 +2679,47 @@
 n seed;%0A        
+      
+%0A        
 %0A        throw n
@@ -2742,21 +2742,16 @@
 ion(%22No 
-live 
 seeds fo

Change logic that chooses job tracker to account for seeds not being alive on startup