Adding basic support for abortable rebalances, more to follow

contrib/ec2-testing/test/voldemort/utils/Ec2RebalanceTest.java

@@ -44,7 +44,7 @@
 import voldemort.client.protocol.RequestFormatType;
 import voldemort.client.protocol.admin.AdminClient;
 import voldemort.client.protocol.admin.AdminClientConfig;
-import voldemort.client.rebalance.AbstractRebalanceTest;
+import voldemort.client.rebalance.AbstractNonZonedRebalanceTest;
 import voldemort.cluster.Cluster;
 import voldemort.cluster.Node;
 import voldemort.server.RequestRoutingType;
@@ -55,7 +55,7 @@
 
 /**
  */
-public class Ec2RebalanceTest extends AbstractRebalanceTest {
+public class Ec2RebalanceTest extends AbstractNonZonedRebalanceTest {
 
     private static int NUM_KEYS;
 
@@ -66,7 +66,10 @@ public class Ec2RebalanceTest extends AbstractRebalanceTest {
 
     private Map<Integer, String> nodeIdsInv = new HashMap<Integer, String>();
     private List<String> activeHostNames = new ArrayList<String>();
-    private boolean useDonorBased = true;
+
+    public Ec2RebalanceTest() {
+        super(true, true);
+    }
 
     @BeforeClass
     public static void ec2Setup() throws Exception {
@@ -209,11 +212,6 @@ protected void stopServer(List<Integer> nodesToStop) throws Exception {
         stopCluster(hostsToStop, ec2RebalanceTestConfig);
     }
 
-    @Override
-    protected boolean useDonorBased() {
-        return this.useDonorBased;
-    }
-
     private static class Ec2RebalanceTestConfig extends Ec2RemoteTestConfig {
 
         private String configDirName;

src/java/voldemort/client/protocol/admin/AdminClient.java

@@ -73,7 +73,6 @@
 import voldemort.store.StoreUtils;
 import voldemort.store.metadata.MetadataStore;
 import voldemort.store.metadata.MetadataStore.VoldemortState;
-import voldemort.store.mysql.MysqlStorageConfiguration;
 import voldemort.store.readonly.ReadOnlyStorageConfiguration;
 import voldemort.store.readonly.ReadOnlyStorageFormat;
 import voldemort.store.readonly.ReadOnlyStorageMetadata;
@@ -2118,7 +2117,8 @@ public void updateEntries(int nodeId,
         }
 
         /**
-         * Fetch key/value tuples belonging to a node with given key values
+         * Fetch key/value tuples from a given server, directly from storage
+         * engine
          * 
          * <p>
          * Entries are being queried synchronously
@@ -2139,7 +2139,6 @@ public Iterator<QueryKeyResult> queryKeys(int nodeId,
 
             try {
                 store = adminStoreClient.getSocketStore(nodeId, storeName);
-
             } catch(Exception e) {
                 throw new VoldemortException(e);
             }

src/java/voldemort/server/protocol/admin/AsyncOperationService.java

@@ -1,12 +1,12 @@
 /*
  * Copyright 2008-2010 LinkedIn, Inc
- *
+ * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not
  * use this file except in compliance with the License. You may obtain a copy of
  * the License at
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
@@ -16,10 +16,13 @@
 
 package voldemort.server.protocol.admin;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import com.google.common.collect.ImmutableSet;
 import org.apache.log4j.Logger;
 
 import voldemort.VoldemortException;
@@ -29,11 +32,14 @@
 import voldemort.common.service.SchedulerService;
 import voldemort.common.service.ServiceType;
 
+import com.google.common.collect.ImmutableSet;
+
 /**
  * Asynchronous job scheduler for admin service operations.
- *
- * TODO: requesting a unique id, then creating an operation with that id seems like a bad API design.
- *
+ * 
+ * TODO: requesting a unique id, then creating an operation with that id seems
+ * like a bad API design.
+ * 
  */
 @JmxManaged(description = "Asynchronous operation execution")
 public class AsyncOperationService extends AbstractService {
@@ -96,6 +102,19 @@ public String getStatus(int id) {
         }
     }
 
+    public List<Integer> getMatchingAsyncOperationList(String jobDescPattern, boolean showCompleted) {
+        List<Integer> operationIds = getAsyncOperationList(showCompleted);
+        List<Integer> matchingOperationIds = new ArrayList<Integer>(operationIds.size());
+        for(Integer operationId: operationIds) {
+            AsyncOperation operation = operations.get(operationId);
+            String operationDescription = operation.getStatus().getDescription();
+            if(operationDescription != null && operationDescription.indexOf(jobDescPattern) != -1) {
+                matchingOperationIds.add(operationId);
+            }
+        }
+        return matchingOperationIds;
+    }
+
     @JmxOperation(description = "Retrieve all operations")
     public String getAllAsyncOperations() {
         String result;
@@ -108,23 +127,25 @@ public String getAllAsyncOperations() {
     }
 
     /**
-     * Get list of asynchronous operations on this node. By default, only the pending
-     * operations are returned.
+     * Get list of asynchronous operations on this node. By default, only the
+     * pending operations are returned.
+     * 
      * @param showCompleted Show completed operations
      * @return A list of operation ids.
      */
     public List<Integer> getAsyncOperationList(boolean showCompleted) {
         /**
-         * Create a copy using an immutable set to avoid a {@link java.util.ConcurrentModificationException}
+         * Create a copy using an immutable set to avoid a
+         * {@link java.util.ConcurrentModificationException}
          */
         Set<Integer> keySet = ImmutableSet.copyOf(operations.keySet());
 
-        if (showCompleted)
+        if(showCompleted)
             return new ArrayList<Integer>(keySet);
 
         List<Integer> keyList = new ArrayList<Integer>();
-        for (int key: keySet) {
-            if (!operations.get(key).getStatus().isComplete())
+        for(int key: keySet) {
+            if(!operations.get(key).getStatus().isComplete())
                 keyList.add(key);
         }
         return keyList;
@@ -142,7 +163,7 @@ public AsyncOperationStatus getOperationStatus(int requestId) {
     public String stopAsyncOperation(int requestId) {
         try {
             stopOperation(requestId);
-        } catch (VoldemortException e) {
+        } catch(VoldemortException e) {
             return e.getMessage();
         }
 
@@ -158,6 +179,7 @@ public void stopOperation(int requestId) {
 
     /**
      * Generate a unique request id
+     * 
      * @return A new, guaranteed unique, request id
      */
     public int getUniqueRequestId() {

src/java/voldemort/server/storage/RepairJob.java

@@ -97,8 +97,7 @@ public void run() {
                     long repairSlops = 0L;
                     long numDeletedKeys = 0;
                     while(iterator.hasNext()) {
-                        Pair<ByteArray, Versioned<byte[]>> keyAndVal;
-                        keyAndVal = iterator.next();
+                        Pair<ByteArray, Versioned<byte[]>> keyAndVal = iterator.next();
                         List<Node> nodes = routingStrategy.routeRequest(keyAndVal.getFirst().get());
 
                         if(!hasDestination(nodes)) {
@@ -111,7 +110,8 @@ public void run() {
                     }
                     closeIterator(iterator);
                     localStats.put(storeDef.getName(), repairSlops);
-                    logger.info("Completed store " + storeDef.getName());
+                    logger.info("Completed store " + storeDef.getName() + " #Scanned:"
+                                + progress.get() + " #Deleted:" + numDeletedKeys);
                 }
             }
         } catch(Exception e) {

src/java/voldemort/store/rebalancing/RedirectingStore.java

@@ -91,22 +91,6 @@ public boolean getIsRedirectingStoreEnabled() {
         return this.isRedirectingStoreEnabled.get();
     }
 
-    @Override
-    public void put(ByteArray key, Versioned<byte[]> value, byte[] transforms)
-            throws VoldemortException {
-        RebalancePartitionsInfo stealInfo = redirectingKey(key);
-
-        /**
-         * If I am rebalancing for this key, try to do remote get() , put it
-         * locally first to get the correct version ignoring any
-         * {@link ObsoleteVersionException}
-         */
-        if(stealInfo != null)
-            proxyGetAndLocalPut(key, stealInfo.getDonorId(), transforms);
-
-        getInnerStore().put(key, value, transforms);
-    }
-
     private RebalancePartitionsInfo redirectingKey(ByteArray key) {
         if(VoldemortState.REBALANCING_MASTER_SERVER.equals(metadata.getServerState())
            && isRedirectingStoreEnabled.get()) {
@@ -129,6 +113,9 @@ public List<Versioned<byte[]>> get(ByteArray key, byte[] transforms) throws Vold
          * locally first to get the correct version ignoring any
          * {@link ObsoleteVersionException}
          */
+        // TODO this is correct per se. But, had a heavy performance hit for
+        // cross zone moves. Check locally first, if you cannot find, then go
+        // remote and store it locally.
         if(stealInfo != null) {
             proxyGetAndLocalPut(key, stealInfo.getDonorId(), transforms);
         }
@@ -145,6 +132,9 @@ public List<Version> getVersions(ByteArray key) {
          * locally first to get the correct version ignoring any
          * {@link ObsoleteVersionException}.
          */
+        // TODO this is correct per se. But, had a heavy performance hit for
+        // cross zone moves. Check locally first, if you cannot find, then go
+        // remote and store it locally.
         if(stealInfo != null) {
             proxyGetAndLocalPut(key, stealInfo.getDonorId(), null);
         }
@@ -163,14 +153,37 @@ public Map<ByteArray, List<Versioned<byte[]>>> getAll(Iterable<ByteArray> keys,
                 rebalancePartitionsInfoPerKey.put(key, info);
             }
         }
-
+        // TODO this is correct per se. But, had a heavy performance hit for
+        // cross zone moves. Check locally first, if you cannot find, then go
+        // remote and store it locally.
         if(!rebalancePartitionsInfoPerKey.isEmpty()) {
             proxyGetAllAndLocalPut(rebalancePartitionsInfoPerKey, transforms);
         }
 
         return getInnerStore().getAll(keys, transforms);
     }
 
+    @Override
+    public void put(ByteArray key, Versioned<byte[]> value, byte[] transforms)
+            throws VoldemortException {
+        RebalancePartitionsInfo stealInfo = redirectingKey(key);
+
+        /**
+         * If I am rebalancing for this key, try to do remote get() , put it
+         * locally first to get the correct version ignoring any
+         * {@link ObsoleteVersionException}
+         */
+        if(stealInfo != null)
+            proxyGetAndLocalPut(key, stealInfo.getDonorId(), transforms);
+
+        // Just sychronous replication,if remote fails, I fail.
+        if(stealInfo != null)
+            proxyPut(key, value, transforms, stealInfo.getDonorId());
+        // TODO if I fail though for some reason, the aborting rebalance can
+        // surface phantom writes
+        getInnerStore().put(key, value, transforms);
+    }
+
     /**
      * TODO : Handle delete correctly.
      * <p>
@@ -219,6 +232,31 @@ private List<Versioned<byte[]>> proxyGet(ByteArray key, int donorNodeId, byte[]
         }
     }
 
+    /**
+     * Replay the put to a remote proxy node so we will have the data available
+     * at the proxy host, in case the rebalancing fails
+     * 
+     * @param key
+     * @param value
+     * @param transforms
+     * @param donorNodeId
+     * @throws ProxyUnreachableException if donor node can't be reached
+     */
+    private void proxyPut(ByteArray key, Versioned<byte[]> value, byte[] transforms, int donorNodeId) {
+        Node donorNode = metadata.getCluster().getNodeById(donorNodeId);
+        checkNodeAvailable(donorNode);
+        long startNs = System.nanoTime();
+        try {
+            Store<ByteArray, byte[], byte[]> redirectingStore = getRedirectingSocketStore(getName(),
+                                                                                          donorNodeId);
+            redirectingStore.put(key, value, transforms);
+            recordSuccess(donorNode, startNs);
+        } catch(UnreachableStoreException e) {
+            recordException(donorNode, startNs, e);
+            throw new ProxyUnreachableException("Failed to reach proxy node " + donorNode, e);
+        }
+    }
+
     private void checkNodeAvailable(Node donorNode) {
         if(!failureDetector.isAvailable(donorNode))
             throw new ProxyUnreachableException("Failed to reach proxy node " + donorNode

src/java/voldemort/store/socket/clientrequest/ClientRequestExecutorPool.java

@@ -131,6 +131,17 @@ public ClientRequestExecutorFactory getFactory() {
         return factory;
     }
 
+    /***
+     * Create a new socket store to talk to a given server for a specific store
+     * 
+     * Note: IGNORE_CHECKS will only be honored for Protobuf request format
+     * 
+     * @param storeName
+     * @param hostName
+     * @param port
+     * @param requestFormatType protocol to use
+     * @param requestRoutingType routed/ignore checks/normal
+     */
     @Override
     public SocketStore create(String storeName,
                               String hostName,

src/java/voldemort/utils/ByteArray.java

@@ -41,7 +41,7 @@ public boolean equals(Object obj) {
 
     @Override
     public String toString() {
-        return Arrays.toString(underlying);
+        return ByteUtils.toHexString(underlying);
     }
 
     /**