[BZ 185375] add failure detection

There is a now a failure threshold that if exceeded will cause the migration to
be aborted. The rationale is that the read/write rates will likely have to be
adjusted based on each environment. If we are reading too fast for example,
and start generating lots of failiures, it makes sense to go ahead and
terminate the migration and restart with a lower read rate.

modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/MigrateAggregateMetrics.java

@@ -112,7 +112,10 @@ public String toString() {
 
     private MetricsRegistry metricsRegistry;
 
-    @Override
+    private int totalMetrics;
+
+    private double failureThreshold = Double.parseDouble(System.getProperty("rhq.storage.failure-threshold", "0.05"));
+
     public void setSession(Session session) {
         this.session = session;
     }
@@ -179,6 +182,7 @@ public void execute() {
         } finally {
             stopwatch.stop();
             log.info("Finished data migration in " + stopwatch.elapsed(TimeUnit.SECONDS) + " sec");
+            shutdown();
         }
     }
 
@@ -187,7 +191,7 @@ private int getWriteLimit() {
     }
 
     private int getReadLimit() {
-        return Integer.parseInt(System.getProperty("rhq.storage.request.read-limit", "100"));
+        return Integer.parseInt(System.getProperty("rhq.storage.request.read-limit", "110"));
     }
 
     private int getNumberOfUpNodes() {
@@ -200,6 +204,17 @@ private int getNumberOfUpNodes() {
         return count;
     }
 
+    private void shutdown() {
+        try {
+            log.info("Shutting down migration thread pools...");
+            scheduler.shutdown();
+            threadPool.shutdown();
+            scheduler.awaitTermination(5, TimeUnit.SECONDS);
+            threadPool.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+        }
+    }
+
     private Keyspace createKeyspace() {
         CassandraHostConfigurator configurator = new CassandraHostConfigurator("127.0.01");
         Map<String, String> credentials = ImmutableMap.of(
@@ -264,7 +279,6 @@ private void scheduleMigrations(Iterator<Integer> keyIterator, final PreparedSta
                 log.debug("Scheduling " + bucket + " data migration for schedule id " + scheduleId);
             }
             migrations.addTask();
-//            migrationsQeue.offer(new MigrationArgs(scheduleId, bucket, query, migratedScheduleIds, migrationLog));
             scheduler.submit(new Runnable() {
                 @Override
                 public void run() {
@@ -283,7 +297,7 @@ public void run() {
                 }
             });
         }
-
+        totalMetrics += count;
         stopwatch.stop();
         log.info("Finished scanning for schedule ids with " + bucket + " data in " +
             stopwatch.elapsed(TimeUnit.SECONDS) + " sec");
@@ -347,6 +361,12 @@ public void run() {
                     } else {
                         reportMigrationRates = true;
                     }
+
+                    if ((failedMigrations.get() / totalMetrics) > failureThreshold) {
+                        migrations.abort("The failure threshold has been exceeded with " + failedMigrations.get() +
+                            " failed migrations");
+                    }
+
                     Thread.sleep(30000);
                 }
             } catch (InterruptedException e) {