added configurable schema agreement settings for issue hector-client#302

core/src/main/java/me/prettyprint/cassandra/service/AbstractCluster.java

@@ -48,8 +48,9 @@ public abstract class AbstractCluster implements Cluster {
   /**
    * Linked to Cassandra StorageProxy.
    */
-  private static final int RING_DELAY = 30 * 1000; // delay after which we assume ring has stablized
-
+  public static final int RING_DELAY_DEF = 30 * 1000; // delay after which we assume ring has stablized
+  public static final int WAIT_FOR_SCHEMA_AGREEMENT_SLEEP_TIME = 1000;
+
   protected final HConnectionManager connectionManager;
   private final String name;
   private final CassandraHostConfigurator configurator;
@@ -60,6 +61,9 @@ public abstract class AbstractCluster implements Cluster {
   private Set<CassandraHost> knownPoolHosts;
   protected final ExceptionsTranslator xtrans;
   private final Map<String, String> credentials;
+  private final int ringDelay;
+  private final int schemaAgreementSleepTime;
+
 
   public AbstractCluster(String clusterName, CassandraHostConfigurator cassandraHostConfigurator) {
     this(clusterName, cassandraHostConfigurator, EMPTY_CREDENTIALS);
@@ -74,6 +78,8 @@ public AbstractCluster(String clusterName, CassandraHostConfigurator cassandraHo
     xtrans = new ExceptionsTranslatorImpl();
     clockResolution = cassandraHostConfigurator.getClockResolution();
     this.credentials = Collections.unmodifiableMap(credentials);
+    ringDelay = cassandraHostConfigurator.getRingDelay();
+    schemaAgreementSleepTime = cassandraHostConfigurator.getSchemaAgreementSleepTime();
   }
 
   @Override
@@ -303,7 +309,7 @@ public Void execute(Cassandra.Client cassandra) throws HectorException {
   }
 
 
-  protected static void waitForSchemaAgreement(Cassandra.Client cassandra) throws InvalidRequestException, TException, InterruptedException {
+  protected void waitForSchemaAgreement(Cassandra.Client cassandra) throws InvalidRequestException, TException, InterruptedException {
     int waited = 0;
     int versions = 0;
     while (versions != 1) {
@@ -315,10 +321,10 @@ protected static void waitForSchemaAgreement(Cassandra.Client cassandra) throws
       }
 
       if (versions != 1) {
-        Thread.sleep(1000);
-        waited += 1000;
-        if (waited > RING_DELAY)
-          throw new RuntimeException("Could not reach schema agreement in " + RING_DELAY + "ms");
+        Thread.sleep(schemaAgreementSleepTime);
+        waited += schemaAgreementSleepTime;
+        if (waited > ringDelay)
+          throw new RuntimeException("Could not reach schema agreement in " + ringDelay + "ms");
       }
     }
   }

core/src/main/java/me/prettyprint/cassandra/service/CassandraHostConfigurator.java

@@ -44,6 +44,8 @@ public final class CassandraHostConfigurator implements Serializable {
   private boolean runAutoDiscoveryAtStartup = false;
   private boolean useSocketKeepalive = false;
   private HOpTimer opTimer = new NullOpTimer();
+  private int ringDelay = AbstractCluster.RING_DELAY_DEF;
+  private int schemaAgreementSleepTime = AbstractCluster.WAIT_FOR_SCHEMA_AGREEMENT_SLEEP_TIME;
 
   public CassandraHostConfigurator() {
     this.hosts = null;
@@ -316,6 +318,30 @@ public void setUseSocketKeepalive(boolean useSocketKeepalive) {
     this.useSocketKeepalive = useSocketKeepalive;
   }
 
+  /**
+   * Delay in ms for which we will assume the ring has stabilized after a schema modification
+   * @return
+   */
+  public int getRingDelay() {
+    return ringDelay;
+  }
+
+  public void setRingDelay(int ringDelay) {
+    this.ringDelay = ringDelay;
+  }
+
+  /**
+   * The amount of time in ms to wait for schema agreements
+   * @return
+   */
+  public int getSchemaAgreementSleepTime() {
+    return schemaAgreementSleepTime;
+  }
+
+  public void setSchemaAgreementSleepTime(int schemaAgreementSleepTime) {
+    this.schemaAgreementSleepTime = schemaAgreementSleepTime;
+  }
+
 
 
 }