Merge commit 'db6ac447895255c84ef0e0cbd0303ffa6b45e05b'

beobal · Apr 17, 2013 · 266221c · 266221c
2 parents 90c0b5f + db6ac44
commit 266221c
Show file tree

Hide file tree

Showing 12 changed files with 2,153 additions and 283 deletions.
diff --git a/bin/run-class.sh b/bin/run-class.sh
@@ -21,8 +21,14 @@ if [ $# -lt 1 ]; then
 	exit 1
 fi
 
-script_path=$(readlink -f "$0")
-script_dir=`dirname "$script_path"`
+if [ $(uname) == 'Darwin' ]; then
+    pushd `dirname $0` > /dev/null
+    script_dir=$(pwd)
+    popd > /dev/null
+else
+    script_path=$(readlink -f "$0")
+    script_dir=`dirname "$script_path"`
+fi
 
 base_dir=`dirname "$script_dir"`
 

diff --git a/src/java/voldemort/client/protocol/admin/StreamingClient.java b/src/java/voldemort/client/protocol/admin/StreamingClient.java
@@ -105,8 +105,8 @@ public class StreamingClient {
 
     protected EventThrottler throttler;
 
-    AdminClient adminClient;
-    AdminClientConfig adminClientConfig;
+    private AdminClient adminClient;
+    private AdminClientConfig adminClientConfig;
 
     String bootstrapURL;
 
@@ -135,7 +135,12 @@ public StreamingClient(StreamingClientConfig config) {
         this.bootstrapURL = config.getBootstrapURL();
         CHECKPOINT_COMMIT_SIZE = config.getBatchSize();
         THROTTLE_QPS = config.getThrottleQPS();
+        adminClientConfig = new AdminClientConfig();
+        adminClient = new AdminClient(bootstrapURL, adminClientConfig, new ClientConfig());
+    }
 
+    public AdminClient getAdminClient() {
+        return adminClient;
     }
 
     public synchronized void updateThrottleLimit(int throttleQPS) {
@@ -297,8 +302,6 @@ public synchronized void initStreamingSessions(List<String> stores,
                                                    List<Integer> blackListedNodes) {
 
         logger.info("Initializing a streaming session");
-        adminClientConfig = new AdminClientConfig();
-        adminClient = new AdminClient(bootstrapURL, adminClientConfig, new ClientConfig());
         this.checkpointCallback = checkpointCallback;
         this.recoveryCallback = recoveryCallback;
         this.allowMerge = allowMerge;

diff --git a/src/java/voldemort/store/StoreUtils.java b/src/java/voldemort/store/StoreUtils.java
@@ -215,4 +215,14 @@ public static List<String> getStoreNames(List<StoreDefinition> list, boolean ign
                 storeNameSet.add(def.getName());
         return storeNameSet;
     }
+
+    public static HashMap<String, StoreDefinition> getStoreDefsAsMap(List<StoreDefinition> storeDefs) {
+        if(storeDefs == null)
+            return null;
+
+        HashMap<String, StoreDefinition> storeDefMap = new HashMap<String, StoreDefinition>();
+        for(StoreDefinition def: storeDefs)
+            storeDefMap.put(def.getName(), def);
+        return storeDefMap;
+    }
 }
diff --git a/src/java/voldemort/store/routed/PipelineRoutedStore.java b/src/java/voldemort/store/routed/PipelineRoutedStore.java
@@ -29,11 +29,11 @@
 import voldemort.cluster.failuredetector.FailureDetector;
 import voldemort.common.VoldemortOpCode;
 import voldemort.routing.RoutingStrategyType;
+import voldemort.store.CompositeVoldemortRequest;
 import voldemort.store.Store;
 import voldemort.store.StoreDefinition;
 import voldemort.store.StoreRequest;
 import voldemort.store.StoreUtils;
-import voldemort.store.CompositeVoldemortRequest;
 import voldemort.store.nonblockingstore.NonblockingStore;
 import voldemort.store.routed.Pipeline.Event;
 import voldemort.store.routed.Pipeline.Operation;
@@ -75,10 +75,10 @@
  */
 public class PipelineRoutedStore extends RoutedStore {
 
-    private final Map<Integer, NonblockingStore> nonblockingStores;
-    private final Map<Integer, Store<ByteArray, Slop, byte[]>> slopStores;
-    private final Map<Integer, NonblockingStore> nonblockingSlopStores;
-    private final HintedHandoffStrategy handoffStrategy;
+    protected final Map<Integer, NonblockingStore> nonblockingStores;
+    protected final Map<Integer, Store<ByteArray, Slop, byte[]>> slopStores;
+    protected final Map<Integer, NonblockingStore> nonblockingSlopStores;
+    protected final HintedHandoffStrategy handoffStrategy;
     private Zone clientZone;
     private boolean zoneRoutingEnabled;
     private PipelineRoutedStats stats;