kudu-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jdcry...@apache.org
Subject [3/3] kudu git commit: [java client] Few ITClient improvements
Date Thu, 22 Sep 2016 00:33:45 GMT
[java client] Few ITClient improvements

ITClient has been flaky for a while now, mostly due to the "Row count regressed"
issue. I fixed it by using snapshot timestamps, which made me refactor how we build
scanners, which made me add a new counting method in BaseKuduTest.

I continued running the test and saw other issues. Some unchecked errors were not
killing the test, so I added an UncaughtExceptionHandler. I also saw invalid
scanner sequence ID errors that are normal due to how this test runs that were killing
the test. Finally, I converted some plain Exceptions into KuduExceptions which gave
us access to their Status.

Change-Id: I3b5ddca26b66e9fc1f737aaacf98df340f0b9024
Reviewed-on: http://gerrit.cloudera.org:8080/4489
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <adar@cloudera.com>


Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/0ce5ba59
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/0ce5ba59
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/0ce5ba59

Branch: refs/heads/master
Commit: 0ce5ba594412de4365625485ea7b3c1ee21bf28d
Parents: 8fc75a5
Author: Jean-Daniel Cryans <jdcryans@apache.org>
Authored: Tue Sep 20 18:20:01 2016 -0700
Committer: Jean-Daniel Cryans <jdcryans@apache.org>
Committed: Thu Sep 22 00:33:27 2016 +0000

----------------------------------------------------------------------
 .../org/apache/kudu/client/BaseKuduTest.java    | 19 +++--
 .../java/org/apache/kudu/client/ITClient.java   | 75 +++++++++++++++-----
 2 files changed, 68 insertions(+), 26 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kudu/blob/0ce5ba59/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java b/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
index 41015a4..8989f50 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
@@ -140,13 +140,17 @@ public class BaseKuduTest {
       data.addCallbacks(cb, defaultErrorCB);
       data.join(DEFAULT_SLEEP);
     }
-
-    Deferred<RowResultIterator> closer = scanner.close();
-    closer.addCallbacks(cb, defaultErrorCB);
-    closer.join(DEFAULT_SLEEP);
     return counter.get();
   }
 
+  protected static int countRowsInScan(KuduScanner scanner) throws KuduException {
+    int counter = 0;
+    while (scanner.hasMoreRows()) {
+      counter += scanner.nextRows().getNumRows();
+    }
+    return counter;
+  }
+
   /**
    * Scans the table and returns the number of rows.
    * @param table the table
@@ -155,17 +159,12 @@ public class BaseKuduTest {
    */
   protected long countRowsInTable(KuduTable table, KuduPredicate... predicates)
       throws KuduException {
-    long count = 0;
     KuduScanner.KuduScannerBuilder scanBuilder = syncClient.newScannerBuilder(table);
     for (KuduPredicate predicate : predicates) {
       scanBuilder.addPredicate(predicate);
     }
     scanBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
-    KuduScanner scanner = scanBuilder.build();
-    while (scanner.hasMoreRows()) {
-      count += scanner.nextRows().getNumRows();
-    }
-    return count;
+    return countRowsInScan(scanBuilder.build());
   }
 
   protected List<String> scanTableToStrings(KuduTable table,

http://git-wip-us.apache.org/repos/asf/kudu/blob/0ce5ba59/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
index 10cf231..c482a86 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
@@ -60,6 +60,8 @@ public class ITClient extends BaseKuduTest {
   private static KuduTable table;
   private static long runtimeInSeconds;
 
+  private volatile long sharedWriteTimestamp;
+
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
 
@@ -89,6 +91,9 @@ public class ITClient extends BaseKuduTest {
 
   @Test(timeout = TEST_TIMEOUT_SECONDS)
   public void test() throws Exception {
+
+    UncaughtExceptionHandler uncaughtExceptionHandler = new UncaughtExceptionHandler();
+
     ArrayList<Thread> threads = new ArrayList<>();
     Thread chaosThread = new Thread(new ChaosThread());
     Thread writerThread = new Thread(new WriterThread());
@@ -99,6 +104,7 @@ public class ITClient extends BaseKuduTest {
     threads.add(scannerThread);
 
     for (Thread thread : threads) {
+      thread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
       thread.start();
     }
 
@@ -235,6 +241,7 @@ public class ITClient extends BaseKuduTest {
 
     @Override
     public void run() {
+      session.setExternalConsistencyMode(ExternalConsistencyMode.CLIENT_PROPAGATED);
       while (KEEP_RUNNING_LATCH.getCount() > 0) {
         try {
           OperationResponse resp = session.apply(createBasicSchemaInsert(table, currentRowKey));
@@ -279,6 +286,13 @@ public class ITClient extends BaseKuduTest {
             " returned this error: " + resp.getRowError(), null);
         return true;
       }
+
+      if (resp == null) {
+        return false;
+      }
+
+      sharedWriteTimestamp = resp.getWriteTimestampRaw();
+
       return false;
     }
   }
@@ -299,8 +313,11 @@ public class ITClient extends BaseKuduTest {
 
         boolean shouldContinue;
 
-        // Always scan until we find rows.
-        if (lastRowCount == 0 || random.nextBoolean()) {
+        // First check if we've written at least one row.
+        if (sharedWriteTimestamp == 0) {
+          shouldContinue = true;
+        } else if (lastRowCount == 0 || // Need to full scan once before random reading
+            random.nextBoolean()) {
           shouldContinue = fullScan();
         } else {
           shouldContinue = randomGet();
@@ -322,14 +339,16 @@ public class ITClient extends BaseKuduTest {
     }
 
     /**
-     * Reads a row at random that it knows to exist (smaller than lastRowCount).
-     * @return
+     * Reads a row at random that should exist (smaller than lastRowCount).
+     * @return true if the get was successful, false if there was an error
      */
     private boolean randomGet() {
       int key = random.nextInt(lastRowCount);
       KuduPredicate predicate = KuduPredicate.newComparisonPredicate(
           table.getSchema().getColumnByIndex(0), KuduPredicate.ComparisonOp.EQUAL, key);
-      KuduScanner scanner = localClient.newScannerBuilder(table).addPredicate(predicate).build();
+      KuduScanner scanner = getScannerBuilder()
+          .addPredicate(predicate)
+          .build();
 
       List<RowResult> results = new ArrayList<>();
       while (scanner.hasMoreRows()) {
@@ -338,7 +357,7 @@ public class ITClient extends BaseKuduTest {
           for (RowResult row : ite) {
             results.add(row);
           }
-        } catch (Exception e) {
+        } catch (KuduException e) {
           return checkAndReportError("Got error while getting row " + key, e);
         }
       }
@@ -357,25 +376,33 @@ public class ITClient extends BaseKuduTest {
     }
 
     /**
-     * Rusn a full table scan and updates the lastRowCount.
-     * @return
+     * Runs a full table scan and updates the lastRowCount.
+     * @return true if the full scan was successful, false if there was an error
      */
     private boolean fullScan() {
-      AsyncKuduScanner scannerBuilder = localAsyncClient.newScannerBuilder(table).build();
+      KuduScanner scanner = getScannerBuilder().build();
       try {
-        int rowCount = countRowsInScan(scannerBuilder);
+        int rowCount = countRowsInScan(scanner);
         if (rowCount < lastRowCount) {
           reportError("Row count regressed: " + rowCount + " < " + lastRowCount, null);
           return false;
         }
-        lastRowCount = rowCount;
-        LOG.info("New row count {}", lastRowCount);
-      } catch (Exception e) {
-        checkAndReportError("Got error while row counting", e);
+        if (rowCount > lastRowCount) {
+          lastRowCount = rowCount;
+          LOG.info("New row count {}", lastRowCount);
+        }
+      } catch (KuduException e) {
+        return checkAndReportError("Got error while row counting", e);
       }
       return true;
     }
 
+    private KuduScanner.KuduScannerBuilder getScannerBuilder() {
+      return localClient.newScannerBuilder(table)
+          .readMode(AsyncKuduScanner.ReadMode.READ_AT_SNAPSHOT)
+          .snapshotTimestampRaw(sharedWriteTimestamp);
+    }
+
     /**
      * Checks the passed exception contains "Scanner not found". If it does then it returns
true,
      * else it reports the error and returns false.
@@ -384,12 +411,28 @@ public class ITClient extends BaseKuduTest {
      * @param e the exception to check
      * @return true if the scanner failed because it wasn't false, otherwise false
      */
-    private boolean checkAndReportError(String message, Exception e) {
-      if (!e.getCause().getMessage().contains("Scanner not found")) {
+    private boolean checkAndReportError(String message, KuduException e) {
+      // Do nasty things, expect nasty results. The scanners are a bit too happy to retry
TS
+      // disconnections so we might end up retrying a scanner on a node that restarted, or
we might
+      // get disconnected just after sending an RPC so when we reconnect to the same TS we
might get
+      // the "Invalid call sequence ID" message.
+      if (!e.getStatus().isNotFound() &&
+          !e.getStatus().getMessage().contains("Invalid call sequence ID")) {
         reportError(message, e);
         return false;
       }
       return true;
     }
   }
+
+  class UncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
+
+    @Override
+    public void uncaughtException(Thread t, Throwable e) {
+      // Only report an error if we're still running, else we'll spam the log.
+      if (KEEP_RUNNING_LATCH.getCount() != 0) {
+        reportError("Uncaught exception", new Exception(e));
+      }
+    }
+  }
 }
\ No newline at end of file


Mime
View raw message