Count the connection failure as the condition of quarantine

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java

@@ -140,6 +140,7 @@ public class ClientConfiguration extends AbstractConfiguration<ClientConfigurati
     protected static final String BOOKIE_ERROR_THRESHOLD_PER_INTERVAL = "bookieErrorThresholdPerInterval";
     protected static final String BOOKIE_QUARANTINE_TIME_SECONDS = "bookieQuarantineTimeSeconds";
     protected static final String BOOKIE_QUARANTINE_RATIO = "bookieQuarantineRatio";
+    protected static final String BOOKIE_CONNECTION_ERROR_QUARANTINE_ENABLED = "bookieConnectionErrorQuarantineEnabled";
 
     // Bookie info poll interval
     protected static final String DISK_WEIGHT_BASED_PLACEMENT_ENABLED = "diskWeightBasedPlacementEnabled";
@@ -1456,6 +1457,28 @@ public ClientConfiguration setBookieErrorThresholdPerInterval(long thresholdPerI
         return this;
     }
 
+
+    /**
+     * Set if count the bookie connecting error into the quarantine condition. If this is enabled, the connection
+     * error will be counted into the BookieErrorThresholdPerInterval. So be careful to set the quarantine time.
+     *
+     * @param enabled
+     * @return
+     */
+    public ClientConfiguration setBookieConnectionErrorQuarantineEnabled(boolean enabled) {
+        setProperty(BOOKIE_CONNECTION_ERROR_QUARANTINE_ENABLED, enabled);
+        return this;
+    }
+
+    /**
+     * Get if count the bookie connecting error into the quarantine condition.
+     *
+     * @return
+     */
+    public boolean getBookieConnectionErrorQuarantineEnabled() {
+        return getBoolean(BOOKIE_CONNECTION_ERROR_QUARANTINE_ENABLED, false);
+    }
+
     /**
      * Get the time for which a bookie will be quarantined.
      *

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java

@@ -1818,6 +1818,9 @@ public void operationComplete(ChannelFuture future) {
                     if (state != ConnectionState.CLOSED) {
                         state = ConnectionState.DISCONNECTED;
                     }
+                    if (conf.getBookieConnectionErrorQuarantineEnabled()) {
+                        recordError();
+                    }
                     failedConnectionCounter.inc();
                 }