diff --git a/CHANGES.txt b/CHANGES.txt
index 47307634dc..a13872d5dd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -21,6 +21,7 @@
  * Fix using non-utf8-aware comparison as a sanity check.
    (CASSANDRA-493)
  * Improve default garbage collector options (CASSANDRA-504)
+ * Add "nodeprobe flush" (CASSANDRA-505)
 
 
 0.4.1
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 46399a5fd4..2b51410107 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -613,7 +613,13 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
             logger_.debug("Cleared out all snapshot directories");
     }
 
-    public void forceTableFlushBinary(String tableName) throws IOException
+    /**
+     * Flush all memtables for a table and column families.
+     * @param tableName
+     * @param columnFamilies
+     * @throws IOException
+     */
+    public void forceTableFlush(String tableName, String... columnFamilies) throws IOException
     {
         if (DatabaseDescriptor.getTable(tableName) == null)
         {
@@ -621,16 +627,33 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
         }
 
         Table table = Table.open(tableName);
-        Set<String> columnFamilies = table.getColumnFamilies();
+        Set<String> positiveColumnFamilies = table.getColumnFamilies();
+
+        // no columnFamilies means flush'em all.
+        if (columnFamilies == null || columnFamilies.length == 0)
+        {
+            columnFamilies = positiveColumnFamilies.toArray(new String[positiveColumnFamilies.size()]);
+        }
+
         for (String columnFamily : columnFamilies)
         {
-            ColumnFamilyStore cfStore = table.getColumnFamilyStore(columnFamily);
-            logger_.debug("Forcing flush on keyspace " + tableName + " on CF " + columnFamily);
-            cfStore.forceFlushBinary();
+
+            if (positiveColumnFamilies.contains(columnFamily))
+            {
+                ColumnFamilyStore cfStore = table.getColumnFamilyStore(columnFamily);
+                logger_.debug("Forcing binary flush on keyspace " + tableName + ", CF " + columnFamily);
+                cfStore.forceFlushBinary();
+                logger_.debug("Forcing flush on keyspace " + tableName + ", CF " + columnFamily);
+                cfStore.forceFlush();
+            }
+            else
+            {
+                // this means there was a cf passed in that is not recognized in the keyspace. report it and continue.
+                logger_.warn(String.format("Invalid column family specified: %s. Proceeding with others.", columnFamily));
+            }
         }
     }
 
-
     /* End of MBean interface methods */
     
     /**
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index d2421eaba4..47b85fb875 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -105,11 +105,13 @@ public interface StorageServiceMBean
     public void clearSnapshot() throws IOException;
 
     /**
-     * Flush all binary memtables for a table
+     * Flush all memtables for the given column families, or all columnfamilies for the given table
+     * if none are explicitly listed.
      * @param tableName
+     * @param columnFamilies
      * @throws IOException
      */
-    public void forceTableFlushBinary(String tableName) throws IOException;
+    public void forceTableFlush(String tableName, String... columnFamilies) throws IOException;
 
     /** set the logging level at runtime */
     public void setLog4jLevel(String classQualifier, String level);
diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java
index 43f2065761..de4727feb2 100644
--- a/src/java/org/apache/cassandra/tools/NodeProbe.java
+++ b/src/java/org/apache/cassandra/tools/NodeProbe.java
@@ -183,9 +183,9 @@ public class NodeProbe
         ssProxy.forceTableCompaction();
     }
 
-    public void forceTableFlushBinary(String tableName) throws IOException
+    public void forceTableFlush(String tableName, String... columnFamilies) throws IOException
     {
-        ssProxy.forceTableFlushBinary(tableName);
+        ssProxy.forceTableFlush(tableName, columnFamilies);
     }
 
     /**
@@ -474,7 +474,7 @@ public class NodeProbe
     {
         HelpFormatter hf = new HelpFormatter();
         String header = String.format(
-                "%nAvailable commands: ring, info, cleanup, compact, cfstats, snapshot [name], clearsnapshot, tpstats, flush_binary, " +
+                "%nAvailable commands: ring, info, cleanup, compact, cfstats, snapshot [name], clearsnapshot, tpstats, flush, " +
                 " getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold])");
         String usage = String.format("java %s -host <arg> <command>%n", NodeProbe.class.getName());
         hf.printHelp(usage, "", options, header);
@@ -550,7 +550,7 @@ public class NodeProbe
         {
             probe.printThreadPoolStats(System.out);
         }
-        else if (cmdName.equals("flush_binary"))
+        else if (cmdName.equals("flush"))
         {
             if (probe.getArgs().length < 2)
             {
@@ -558,7 +558,13 @@ public class NodeProbe
                 NodeProbe.printUsage();
                 System.exit(1);
             }
-            probe.forceTableFlushBinary(probe.getArgs()[1]);
+
+            String[] columnFamilies = new String[probe.getArgs().length - 2];
+            for (int i = 0; i < columnFamilies.length; i++)
+            {
+                columnFamilies[i] = probe.getArgs()[i + 2];
+            }   
+            probe.forceTableFlush(probe.getArgs()[1], columnFamilies);
         }
         else if (cmdName.equals("getcompactionthreshold"))
         {