package voldemort.store.readonly.swapper;

import java.util.List;
import java.util.Map;

import voldemort.client.protocol.admin.AdminClient;
import voldemort.cluster.Node;
import voldemort.store.quota.QuotaExceededException;
import voldemort.store.readonly.swapper.AdminStoreSwapper.Response;

import com.google.common.collect.Lists;

/**
*
*/
public class DisableStoreOnFailedNodeFailedFetchStrategy extends FailedFetchStrategy {
    private final String extraInfo;

    public DisableStoreOnFailedNodeFailedFetchStrategy(AdminClient adminClient,
                                                       String extraInfo) {
        super(adminClient);
        this.extraInfo = extraInfo;
    }

    @Override
    protected boolean dealWithIt(String storeName,
                                 long pushVersion,
                                 Map<Node, AdminStoreSwapper.Response> fetchResponseMap)
            throws Exception {
        int numQuotaExceptions = 0;
        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
                if(response.getException() instanceof QuotaExceededException) {
                    numQuotaExceptions++;
                } else {
                    failedNodes.add(entry.getKey().getId());
                }
            }
        }
        // QuotaException trumps all others
        if(numQuotaExceptions > 0) {
            logger.error("We cannot use "
                         + getClass().getSimpleName()
                         + " because there are QuotaExceededExceptions that caused fetch failures...");
            return false;
        }
        return adminClient.readonlyOps.handleFailedFetch(failedNodes,
                                                         storeName,
                                                         pushVersion,
                                                         extraInfo);
    }
}

@@ -36,16 +36,97 @@
 apper;%0A%0A
+import java.io.IOException;%0Aimport java.util.ArrayList;%0Aimport java.util.Arrays;%0A
 import j
@@ -273,27 +273,24 @@
 ore.
-quota.QuotaExceeded
+UnreachableStore
 Exce
@@ -363,16 +363,55 @@
 esponse;
+%0Aimport voldemort.utils.ExceptionUtils;
 %0A%0Aimport
@@ -1052,34 +1052,66 @@
     
-int numQuotaExceptions = 0
+List%3CString%3E nonConnectionErrors = new ArrayList%3CString%3E()
 ;%0A  
@@ -1506,19 +1506,31 @@
         
-if(
+Exception ex = 
 response
@@ -1548,33 +1548,110 @@
 on()
- instanceof QuotaExceeded
+;%0A                boolean connectionFailure = ExceptionUtils.recursiveClassEquals(ex, UnreachableStore
 Exce
@@ -1655,19 +1655,24 @@
 xception
-) %7B
+.class);
 %0A       
@@ -1684,20 +1684,68 @@
     
-    numQuota
+boolean ioError = ExceptionUtils.recursiveClassEquals(ex, IO
 Exce
@@ -1753,11 +1753,15 @@
 tion
-s++
+.class)
 ;%0A  
@@ -1778,75 +1778,132 @@
     
-%7D else %7B%0A                    failedNodes.add(entry.getKey()
+Node node = entry.getKey();%0A                if(connectionFailure %7C%7C ioError) %7B%0A                    int nodeId = node
 .getId()
 );%0A 
@@ -1898,17 +1898,16 @@
 .getId()
-)
 ;%0A      
@@ -1908,34 +1908,63 @@
                 
-%7D%0A
+    failedNodes.add(nodeId);%0A  
             %7D%0A  
@@ -1951,33 +1951,42 @@
 ;%0A              
-%7D
+  %7D else %7B
 %0A        %7D%0A     
@@ -1978,26 +1978,24 @@
         
-%7D%0A
         
 // Quota
@@ -1990,123 +1990,102 @@
     
-// QuotaException trumps all others%0A        if(numQuotaExceptions %3E 0) %7B%0A            logger.error(%22We cannot use %22%0A
+    String nodeErrorMessage = node.briefToString() + %22 threw exception %22%0A                     
     
@@ -2107,16 +2107,19 @@
       + 
+ex.
 getClass
@@ -2128,20 +2128,22 @@
 .get
-Simple
 Name()
+ + %22. %22;
 %0A   
@@ -2163,89 +2163,313 @@
     
-     + %22 because there are QuotaExceededExceptions that caused fetch failures...%22
+nonConnectionErrors.add(nodeErrorMessage);%0A                %7D%0A            %7D%0A        %7D%0A%0A        if(nonConnectionErrors.size() %3E 0) %7B%0A            String errorMessage = Arrays.toString(nonConnectionErrors.toArray());%0A            logger.error(%22There were non connection related errors. Details %22 + errorMessage
 );%0A 

Swap Only on IOException,UnreachableStoreException