merge from 0.8

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1160461 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -58,13 +58,18 @@
    on create/update of the ColumnFamily and CQL 'ALTER' statement (CASSANDRA-3036)
  * return an InvalidRequestException if an indexed column is assigned
    a value larger than 64KB (CASSANDRA-3057)
- * fix of numeric-only and string column names handling in CLI "drop index" (CASSANDRA-3054)
+ * fix of numeric-only and string column names handling in CLI "drop index" 
+   (CASSANDRA-3054)
  * prune index scan resultset back to original request for lazy
    resultset expansion case (CASSANDRA-2964)
  * (Hadoop) fail jobs when Cassandra node has failed but TaskTracker
     has not (CASSANDRA-2388)
  * fix dynamic snitch ignoring nodes when read_repair_chance is zero
    (CASSANDRA-2662)
+ * avoid retaining references to dropped CFS objects in 
+   CompactionManager.estimatedCompactions (CASSANDRA-2708)
+ * expose rpc timeouts per host in MessagingServiceMBean (CASSANDRA-2941)
+ * avoid including cwd in classpath for deb and rpm packages (CASSANDRA-2881)
 
 
 0.8.4

debian/cassandra.in.sh

@@ -5,7 +5,11 @@ CASSANDRA_CONF=/etc/cassandra
 CASSANDRA_HOME=/usr/share/cassandra
 
 # The java classpath (required)
-CLASSPATH=$CASSANDRA_CONF
+if [ -n "$CLASSPATH" ]; then
+    CLASSPATH=$CLASSPATH:$CASSANDRA_CONF
+else
+    CLASSPATH=$CASSANDRA_CONF
+fi
 
 for jar in /usr/share/cassandra/lib/*.jar; do
     CLASSPATH=$CLASSPATH:$jar

redhat/cassandra.in.sh

@@ -2,11 +2,16 @@
 # The directory where Cassandra's configs live (required)
 CASSANDRA_CONF=/etc/cassandra/conf
 
+# The java classpath (required)
+if [ -n "$CLASSPATH" ]; then
+	CLASSPATH=$CLASSPATH:$CASSANDRA_CONF
+else
+	CLASSPATH=$CASSANDRA_CONF
+fi
+
 # use JNA if installed in standard location
 [ -r /usr/share/java/jna.jar ] && CLASSPATH="$CLASSPATH:/usr/share/java/jna.jar"
 
-# The java classpath (required)
-CLASSPATH=$CLASSPATH:$CASSANDRA_CONF
 
 for jar in /usr/share/cassandra/lib/*.jar; do
     CLASSPATH=$CLASSPATH:$jar

src/java/org/apache/cassandra/net/MessagingService.java

@@ -29,6 +29,7 @@ import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
@@ -102,6 +103,10 @@ public final class MessagingService implements MessagingServiceMBean
     private final Map<StorageService.Verb, Integer> lastDropped = Collections.synchronizedMap(new EnumMap<StorageService.Verb, Integer>(StorageService.Verb.class));
     private final Map<StorageService.Verb, Integer> lastDroppedInternal = new EnumMap<StorageService.Verb, Integer>(StorageService.Verb.class);
 
+    private long totalTimeouts = 0;
+    private long recentTotalTimeouts = 0;
+    private final Map<String, AtomicLong> timeoutsPerHost = new HashMap<String, AtomicLong>();
+    private final Map<String, AtomicLong> recentTimeoutsPerHost = new HashMap<String, AtomicLong>();
     private final List<ILatencySubscriber> subscribers = new ArrayList<ILatencySubscriber>();
     private static final long DEFAULT_CALLBACK_TIMEOUT = (long) (1.1 * DatabaseDescriptor.getRpcTimeout());
 
@@ -141,6 +146,17 @@ public final class MessagingService implements MessagingServiceMBean
             {
                 Pair<InetAddress, IMessageCallback> expiredValue = pair.right;
                 maybeAddLatency(expiredValue.right, expiredValue.left, (double) DatabaseDescriptor.getRpcTimeout());
+                totalTimeouts++;
+                String ip = expiredValue.left.getHostAddress();
+                AtomicLong c = timeoutsPerHost.get(ip);
+                if (c == null)
+                    c = timeoutsPerHost.put(ip, new AtomicLong());
+                c.incrementAndGet();
+                // we only create AtomicLong instances here, so that the write
+                // access to the hashmap happens single-threadedly.
+                if (recentTimeoutsPerHost.get(ip) == null)
+                    recentTimeoutsPerHost.put(ip, new AtomicLong());
+
                 return null;
             }
         };
@@ -649,4 +665,39 @@ public final class MessagingService implements MessagingServiceMBean
         }
         return map;
     }
+
+    public long getTotalTimeouts()
+    {
+        return totalTimeouts;
+    }
+
+    public long getRecentTotalTimouts()
+    {
+        long recent = totalTimeouts - recentTotalTimeouts;
+        recentTotalTimeouts = totalTimeouts;
+        return recent;
+    }
+
+    public Map<String, Long> getTimeoutsPerHost()
+    {
+        Map<String, Long> result = new HashMap<String, Long>();
+        for (Map.Entry<String, AtomicLong> entry: timeoutsPerHost.entrySet())
+        {
+            result.put(entry.getKey(), entry.getValue().get());
+        }
+        return result;
+    }
+
+    public Map<String, Long> getRecentTimeoutsPerHost()
+    {
+        Map<String, Long> result = new HashMap<String, Long>();
+        for (Map.Entry<String, AtomicLong> entry: recentTimeoutsPerHost.entrySet())
+        {
+            String ip = entry.getKey();
+            AtomicLong recent = entry.getValue();
+            Long timeout = timeoutsPerHost.get(ip).get();
+            result.put(ip, timeout - recent.getAndSet(timeout));
+        }
+        return result;
+    }
 }

src/java/org/apache/cassandra/net/MessagingServiceMBean.java

@@ -59,4 +59,24 @@ public interface MessagingServiceMBean
      * dropped message counts since last called
      */
     public Map<String, Integer> getRecentlyDroppedMessages();
+
+    /**
+     * Total number of timeouts happened on this node
+     */
+    public long getTotalTimeouts();
+
+    /**
+     * Number of timeouts per host
+     */
+    public Map<String, Long> getTimeoutsPerHost();
+
+    /**
+     * Number of timeouts since last check.
+     */
+    public long getRecentTotalTimouts();
+
+    /**
+     * Number of timeouts since last check per host.
+     */
+    public Map<String, Long> getRecentTimeoutsPerHost();
 }

src/java/org/apache/cassandra/tools/NodeCmd.java

@@ -739,11 +739,19 @@ public class NodeCmd
 
     private static void handleSnapshots(NodeCommand nc, String tag, String[] cmdArgs, NodeProbe probe) throws InterruptedException, IOException
     {
-        int length = cmdArgs.length > 1 ? cmdArgs.length - 1 : 0;
-        String[] keyspaces = new String[length];
-        for (int i = 0; i < keyspaces.length; i++)
-            keyspaces[i] = cmdArgs[i + 1];
-
+        String[] keyspaces = Arrays.copyOfRange(cmdArgs, 0, cmdArgs.length);
+        System.out.print("Requested snapshot for: ");        
+        if ( keyspaces.length > 0 ) 
+        {
+          for (int i = 0; i < keyspaces.length; i++) 
+              System.out.print(keyspaces[i] + " ");            
+        } 
+        else 
+        {
+            System.out.print("all keyspaces");
+        }
+        System.out.println();
+        
         switch (nc)
         {
             case SNAPSHOT :