Disabling Fetch Fails BnP HA

When a node is in offline mode, it responds with fetch disabled
error. Currently the error is not treated as a soft error and this
fails the HA BnP.

With this code change, FetchDisabled is considered as a soft error
and the fetch will continue as normal.

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

@@ -1052,7 +1052,7 @@ public VAdminProto.AsyncOperationStatusResponse handleFetchROStore(VAdminProto.F
                                                                                                             .setStatus("started");
         try {
             if(!metadataStore.getReadOnlyFetchEnabledUnlocked()) {
-                throw new VoldemortException("Read-only fetcher is disabled in "
+                throw new ReadOnlyFetchDisabledException("Read-only fetcher is disabled in "
                                              + metadataStore.getServerStateUnlocked()
                                              + " state on node " + metadataStore.getNodeId());
             }

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

@@ -0,0 +1,17 @@
+package voldemort.server.protocol.admin;
+
+import voldemort.VoldemortApplicationException;
+
+/**
+ * This exception is thrown when fetch is disabled explicitly on the voldemort
+ * server you can see the state of the fetcher on a server by running the
+ * command
+ * 
+ * bin/vadmin.sh meta get readonly.fetch.enabled --url <AdminUrl>
+ */
+public class ReadOnlyFetchDisabledException extends VoldemortApplicationException {
+
+    public ReadOnlyFetchDisabledException(String s) {
+        super(s);
+    }
+}

src/java/voldemort/store/ErrorCodeMapper.java

@@ -23,6 +23,7 @@
 import voldemort.VoldemortException;
 import voldemort.VoldemortUnsupportedOperationalException;
 import voldemort.server.protocol.admin.AsyncOperationStoppedException;
+import voldemort.server.protocol.admin.ReadOnlyFetchDisabledException;
 import voldemort.server.rebalance.AlreadyRebalancingException;
 import voldemort.server.rebalance.VoldemortRebalancingException;
 import voldemort.store.quota.QuotaExceededException;
@@ -66,6 +67,7 @@ public ErrorCodeMapper() {
         codeToException.put((short) 18, SlopStreamingDisabledException.class);
         codeToException.put((short) 19, InvalidBootstrapURLException.class);
         codeToException.put((short) 20, AsyncOperationStoppedException.class);
+        codeToException.put((short) 21, ReadOnlyFetchDisabledException.class);
 
         exceptionToCode = new HashMap<Class<? extends VoldemortException>, Short>();
         for(Map.Entry<Short, Class<? extends VoldemortException>> entry: codeToException.entrySet())
@@ -75,7 +77,8 @@ public ErrorCodeMapper() {
     public VoldemortException getError(short code, String message) {
         Class<? extends VoldemortException> klass = codeToException.get(code);
         if(klass == null)
-            return new UnknownFailure(Integer.toString(code));
+            return new UnknownFailure(message + ". Unrecognized error code: "
+                                      + Integer.toString(code));
         else
             return ReflectUtils.callConstructor(klass, new Object[] { message });
     }

src/java/voldemort/store/readonly/swapper/DisableStoreOnFailedNodeFailedFetchStrategy.java

@@ -8,6 +8,7 @@
 
 import voldemort.client.protocol.admin.AdminClient;
 import voldemort.cluster.Node;
+import voldemort.server.protocol.admin.ReadOnlyFetchDisabledException;
 import voldemort.store.UnreachableStoreException;
 import voldemort.store.readonly.swapper.AdminStoreSwapper.Response;
 import voldemort.utils.ExceptionUtils;
@@ -31,30 +32,35 @@ protected boolean dealWithIt(String storeName,
                                  long pushVersion,
                                  Map<Node, AdminStoreSwapper.Response> fetchResponseMap)
             throws Exception {
-        List<String> nonConnectionErrors = new ArrayList<String>();
+        List<String> hardFailures = new ArrayList<String>();
         List<Integer> failedNodes = Lists.newArrayList();
         for(Map.Entry<Node, AdminStoreSwapper.Response> entry: fetchResponseMap.entrySet()) {
-            // Only consider non Quota related exceptions as Failures.
             Response response = entry.getValue();
             if(!response.isSuccessful()) {
-                // Check if there are any exceptions due to Quota
                 Exception ex = response.getException();
-                boolean connectionFailure = ExceptionUtils.recursiveClassEquals(ex, UnreachableStoreException.class);
-                boolean ioError = ExceptionUtils.recursiveClassEquals(ex, IOException.class);
+                Class[] softErrors = {UnreachableStoreException.class , IOException.class , ReadOnlyFetchDisabledException.class };
+                boolean isSoftError = false;
+                for(Class softError : softErrors) {
+                    if(ExceptionUtils.recursiveClassEquals(ex, softError)) {
+                        isSoftError = true;
+                        break;
+                    }
+                }
+
                 Node node = entry.getKey();
-                if(connectionFailure || ioError) {
+                if(isSoftError) {
                     int nodeId = node.getId();
                     failedNodes.add(nodeId);
                 } else {
                     String nodeErrorMessage = node.briefToString() + " threw exception "
                                               + ex.getClass().getName() + ". ";
-                    nonConnectionErrors.add(nodeErrorMessage);
+                    hardFailures.add(nodeErrorMessage);
                 }
             }
         }
 
-        if(nonConnectionErrors.size() > 0) {
-            String errorMessage = Arrays.toString(nonConnectionErrors.toArray());
+        if(hardFailures.size() > 0) {
+            String errorMessage = Arrays.toString(hardFailures.toArray());
             logger.error("There were non connection related errors. Details " + errorMessage);
             return false;
         }