kudu-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mpe...@apache.org
Subject [3/4] kudu git commit: java: Move scan token tests into their own class
Date Tue, 03 Jul 2018 02:59:26 GMT
java: Move scan token tests into their own class

A follow-up commit will add additional tests there.

Change-Id: Iefc23f0193cb24a00005a5554881eedbaac15929
Reviewed-on: http://gerrit.cloudera.org:8080/10724
Tested-by: Kudu Jenkins
Reviewed-by: Dan Burkert <danburkert@apache.org>


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

Branch: refs/heads/master
Commit: 0cda8c8ee9dbd982763c3706a6defde1fffc6d00
Parents: 8155d8c
Author: Mike Percy <mpercy@apache.org>
Authored: Fri Jun 22 19:44:14 2018 -0700
Committer: Mike Percy <mpercy@apache.org>
Committed: Tue Jul 3 02:58:04 2018 +0000

----------------------------------------------------------------------
 .../org/apache/kudu/client/TestKuduClient.java  | 228 ------------------
 .../org/apache/kudu/client/TestScanToken.java   | 235 +++++++++++++++++++
 .../org/apache/kudu/util/ClientTestUtil.java    |  45 ++++
 3 files changed, 280 insertions(+), 228 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kudu/blob/0cda8c8e/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
index 3d84cb1..ce88fd7 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
@@ -49,10 +49,8 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterators;
 import com.stumbleupon.async.Deferred;
 
 import org.apache.kudu.util.TimestampUtil;
@@ -586,238 +584,12 @@ public class TestKuduClient extends BaseKuduTest {
     ).size());
   }
 
-  /**
-   * Counts the rows in the provided scan tokens.
-   */
-  private int countScanTokenRows(List<KuduScanToken> tokens) throws Exception {
-    final AtomicInteger count = new AtomicInteger(0);
-    List<Thread> threads = new ArrayList<>();
-    for (final KuduScanToken token : tokens) {
-      final byte[] serializedToken = token.serialize();
-      Thread thread = new Thread(new Runnable() {
-        @Override
-        public void run() {
-          try (KuduClient contextClient = new KuduClient.KuduClientBuilder(masterAddresses)
-              .defaultAdminOperationTimeoutMs(DEFAULT_SLEEP)
-              .build()) {
-            KuduScanner scanner = KuduScanToken.deserializeIntoScanner(serializedToken, contextClient);
-            try {
-              int localCount = 0;
-              while (scanner.hasMoreRows()) {
-                localCount += Iterators.size(scanner.nextRows());
-              }
-              count.addAndGet(localCount);
-            } finally {
-              scanner.close();
-            }
-          } catch (Exception e) {
-            LOG.error("exception in parallel token scanner", e);
-          }
-        }
-      });
-      thread.run();
-      threads.add(thread);
-    }
-
-    for (Thread thread : threads) {
-      thread.join();
-    }
-    return count.get();
-  }
-
   @Test
   public void testGetAuthnToken() throws Exception {
     byte[] token = client.exportAuthenticationCredentials().join();
     assertNotNull(token);
   }
 
-  /**
-   * Tests scan tokens by creating a set of scan tokens, serializing them, and
-   * then executing them in parallel with separate client instances. This
-   * simulates the normal usecase of scan tokens being created at a central
-   * planner and distributed to remote task executors.
-   */
-  @Test
-  public void testScanTokens() throws Exception {
-    int saveFetchTablets = AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP;
-    try {
-      // For this test, make sure that we cover the case that not all tablets
-      // are returned in a single batch.
-      AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP = 4;
-
-      Schema schema = createManyStringsSchema();
-      CreateTableOptions createOptions = new CreateTableOptions();
-      createOptions.addHashPartitions(ImmutableList.of("key"), 8);
-
-      PartialRow splitRow = schema.newPartialRow();
-      splitRow.addString("key", "key_50");
-      createOptions.addSplitRow(splitRow);
-
-      syncClient.createTable(tableName, schema, createOptions);
-
-      KuduSession session = syncClient.newSession();
-      session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
-      KuduTable table = syncClient.openTable(tableName);
-      for (int i = 0; i < 100; i++) {
-        Insert insert = table.newInsert();
-        PartialRow row = insert.getRow();
-        row.addString("key", String.format("key_%02d", i));
-        row.addString("c1", "c1_" + i);
-        row.addString("c2", "c2_" + i);
-        session.apply(insert);
-      }
-      session.flush();
-
-      KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
-      tokenBuilder.batchSizeBytes(0);
-      tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
-      List<KuduScanToken> tokens = tokenBuilder.build();
-      assertEquals(16, tokens.size());
-
-      // KUDU-1809, with batchSizeBytes configured to '0',
-      // the first call to the tablet server won't return
-      // any data.
-      {
-        KuduScanner scanner = tokens.get(0).intoScanner(syncClient);
-        assertEquals(0, scanner.nextRows().getNumRows());
-      }
-
-      for (KuduScanToken token : tokens) {
-        // Sanity check to make sure the debug printing does not throw.
-        LOG.debug(KuduScanToken.stringifySerializedToken(token.serialize(), syncClient));
-      }
-    } finally {
-      AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP = saveFetchTablets;
-    }
-  }
-
-  /**
-   * Tests scan token creation and execution on a table with non-covering range partitions.
-   */
-  @Test
-  public void testScanTokensNonCoveringRangePartitions() throws Exception {
-    Schema schema = createManyStringsSchema();
-    CreateTableOptions createOptions = new CreateTableOptions();
-    createOptions.addHashPartitions(ImmutableList.of("key"), 2);
-
-    PartialRow lower = schema.newPartialRow();
-    PartialRow upper = schema.newPartialRow();
-    lower.addString("key", "a");
-    upper.addString("key", "f");
-    createOptions.addRangePartition(lower, upper);
-
-    lower = schema.newPartialRow();
-    upper = schema.newPartialRow();
-    lower.addString("key", "h");
-    upper.addString("key", "z");
-    createOptions.addRangePartition(lower, upper);
-
-    PartialRow split = schema.newPartialRow();
-    split.addString("key", "k");
-    createOptions.addSplitRow(split);
-
-    syncClient.createTable(tableName, schema, createOptions);
-
-    KuduSession session = syncClient.newSession();
-    session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
-    KuduTable table = syncClient.openTable(tableName);
-    for (char c = 'a'; c < 'f'; c++) {
-      Insert insert = table.newInsert();
-      PartialRow row = insert.getRow();
-      row.addString("key", "" + c);
-      row.addString("c1", "c1_" + c);
-      row.addString("c2", "c2_" + c);
-      session.apply(insert);
-    }
-    for (char c = 'h'; c < 'z'; c++) {
-      Insert insert = table.newInsert();
-      PartialRow row = insert.getRow();
-      row.addString("key", "" + c);
-      row.addString("c1", "c1_" + c);
-      row.addString("c2", "c2_" + c);
-      session.apply(insert);
-    }
-    session.flush();
-
-    KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
-    tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
-    List<KuduScanToken> tokens = tokenBuilder.build();
-    assertEquals(6, tokens.size());
-    assertEquals('f' - 'a' + 'z' - 'h', countScanTokenRows(tokens));
-
-    for (KuduScanToken token : tokens) {
-      // Sanity check to make sure the debug printing does not throw.
-      LOG.debug(KuduScanToken.stringifySerializedToken(token.serialize(), syncClient));
-    }
-  }
-
-  /**
-   * Tests the results of creating scan tokens, altering the columns being
-   * scanned, and then executing the scan tokens.
-   */
-  @Test
-  public void testScanTokensConcurrentAlterTable() throws Exception {
-    Schema schema = new Schema(ImmutableList.of(
-        new ColumnSchema.ColumnSchemaBuilder("key", Type.INT64).nullable(false).key(true).build(),
-        new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64).nullable(false).key(false).build()
-    ));
-    CreateTableOptions createOptions = new CreateTableOptions();
-    createOptions.setRangePartitionColumns(ImmutableList.<String>of());
-    createOptions.setNumReplicas(1);
-    syncClient.createTable(tableName, schema, createOptions);
-
-    KuduTable table = syncClient.openTable(tableName);
-
-    KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
-    List<KuduScanToken> tokens = tokenBuilder.build();
-    assertEquals(1, tokens.size());
-    KuduScanToken token = tokens.get(0);
-
-    // Drop a column
-    syncClient.alterTable(tableName, new AlterTableOptions().dropColumn("a"));
-    try {
-      token.intoScanner(syncClient);
-      fail();
-    } catch (IllegalArgumentException e) {
-      assertTrue(e.getMessage().contains("Unknown column"));
-    }
-
-    // Add back the column with the wrong type.
-    syncClient.alterTable(
-        tableName,
-        new AlterTableOptions().addColumn(
-            new ColumnSchema.ColumnSchemaBuilder("a", Type.STRING).nullable(true).build()));
-    try {
-      token.intoScanner(syncClient);
-      fail();
-    } catch (IllegalStateException e) {
-      assertTrue(e.getMessage().contains(
-          "invalid type INT64 for column 'a' in scan token, expected: STRING"));
-    }
-
-    // Add the column with the wrong nullability.
-    syncClient.alterTable(
-        tableName,
-        new AlterTableOptions().dropColumn("a")
-                               .addColumn(new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64)
-                                                          .nullable(true).build()));
-    try {
-      token.intoScanner(syncClient);
-      fail();
-    } catch (IllegalStateException e) {
-      assertTrue(e.getMessage().contains(
-          "invalid nullability for column 'a' in scan token, expected: NOT NULL"));
-    }
-
-    // Add the column with the correct type and nullability.
-    syncClient.alterTable(
-        tableName,
-        new AlterTableOptions().dropColumn("a")
-                               .addColumn(new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64)
-                                                          .nullable(false)
-                                                          .defaultValue(0L).build()));
-    token.intoScanner(syncClient);
-  }
 
   /**
    * Counts the rows in a table between two optional bounds.

http://git-wip-us.apache.org/repos/asf/kudu/blob/0cda8c8e/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java
new file mode 100644
index 0000000..1761cf7
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java
@@ -0,0 +1,235 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.kudu.client;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.Schema;
+import org.apache.kudu.Type;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.apache.kudu.util.ClientTestUtil.countScanTokenRows;
+import static org.apache.kudu.util.ClientTestUtil.createManyStringsSchema;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TestScanToken extends BaseKuduTest {
+
+  private String testTableName;
+
+  @Before
+  public void setup() {
+    testTableName = getTestMethodNameWithTimestamp();
+  }
+
+  /**
+   * Tests scan tokens by creating a set of scan tokens, serializing them, and
+   * then executing them in parallel with separate client instances. This
+   * simulates the normal usecase of scan tokens being created at a central
+   * planner and distributed to remote task executors.
+   */
+  @Test
+  public void testScanTokens() throws Exception {
+    int saveFetchTablets = AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP;
+    try {
+      // For this test, make sure that we cover the case that not all tablets
+      // are returned in a single batch.
+      AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP = 4;
+
+      Schema schema = createManyStringsSchema();
+      CreateTableOptions createOptions = new CreateTableOptions();
+      createOptions.addHashPartitions(ImmutableList.of("key"), 8);
+
+      PartialRow splitRow = schema.newPartialRow();
+      splitRow.addString("key", "key_50");
+      createOptions.addSplitRow(splitRow);
+
+      syncClient.createTable(testTableName, schema, createOptions);
+
+      KuduSession session = syncClient.newSession();
+      session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+      KuduTable table = syncClient.openTable(testTableName);
+      for (int i = 0; i < 100; i++) {
+        Insert insert = table.newInsert();
+        PartialRow row = insert.getRow();
+        row.addString("key", String.format("key_%02d", i));
+        row.addString("c1", "c1_" + i);
+        row.addString("c2", "c2_" + i);
+        session.apply(insert);
+      }
+      session.flush();
+
+      KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
+      tokenBuilder.batchSizeBytes(0);
+      tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
+      List<KuduScanToken> tokens = tokenBuilder.build();
+      assertEquals(16, tokens.size());
+
+      // KUDU-1809, with batchSizeBytes configured to '0',
+      // the first call to the tablet server won't return
+      // any data.
+      {
+        KuduScanner scanner = tokens.get(0).intoScanner(syncClient);
+        assertEquals(0, scanner.nextRows().getNumRows());
+      }
+
+      for (KuduScanToken token : tokens) {
+        // Sanity check to make sure the debug printing does not throw.
+        LOG.debug(KuduScanToken.stringifySerializedToken(token.serialize(), syncClient));
+      }
+    } finally {
+      AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP = saveFetchTablets;
+    }
+  }
+
+  /**
+   * Tests scan token creation and execution on a table with non-covering range partitions.
+   */
+  @Test
+  public void testScanTokensNonCoveringRangePartitions() throws Exception {
+    Schema schema = createManyStringsSchema();
+    CreateTableOptions createOptions = new CreateTableOptions();
+    createOptions.addHashPartitions(ImmutableList.of("key"), 2);
+
+    PartialRow lower = schema.newPartialRow();
+    PartialRow upper = schema.newPartialRow();
+    lower.addString("key", "a");
+    upper.addString("key", "f");
+    createOptions.addRangePartition(lower, upper);
+
+    lower = schema.newPartialRow();
+    upper = schema.newPartialRow();
+    lower.addString("key", "h");
+    upper.addString("key", "z");
+    createOptions.addRangePartition(lower, upper);
+
+    PartialRow split = schema.newPartialRow();
+    split.addString("key", "k");
+    createOptions.addSplitRow(split);
+
+    syncClient.createTable(testTableName, schema, createOptions);
+
+    KuduSession session = syncClient.newSession();
+    session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+    KuduTable table = syncClient.openTable(testTableName);
+    for (char c = 'a'; c < 'f'; c++) {
+      Insert insert = table.newInsert();
+      PartialRow row = insert.getRow();
+      row.addString("key", "" + c);
+      row.addString("c1", "c1_" + c);
+      row.addString("c2", "c2_" + c);
+      session.apply(insert);
+    }
+    for (char c = 'h'; c < 'z'; c++) {
+      Insert insert = table.newInsert();
+      PartialRow row = insert.getRow();
+      row.addString("key", "" + c);
+      row.addString("c1", "c1_" + c);
+      row.addString("c2", "c2_" + c);
+      session.apply(insert);
+    }
+    session.flush();
+
+    KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
+    tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
+    List<KuduScanToken> tokens = tokenBuilder.build();
+    assertEquals(6, tokens.size());
+    assertEquals('f' - 'a' + 'z' - 'h',
+                 countScanTokenRows(tokens,
+                                    syncClient.getMasterAddressesAsString(),
+                                    syncClient.getDefaultOperationTimeoutMs()));
+
+    for (KuduScanToken token : tokens) {
+      // Sanity check to make sure the debug printing does not throw.
+      LOG.debug(KuduScanToken.stringifySerializedToken(token.serialize(), syncClient));
+    }
+  }
+
+  /**
+   * Tests the results of creating scan tokens, altering the columns being
+   * scanned, and then executing the scan tokens.
+   */
+  @Test
+  public void testScanTokensConcurrentAlterTable() throws Exception {
+    Schema schema = new Schema(ImmutableList.of(
+        new ColumnSchema.ColumnSchemaBuilder("key", Type.INT64).nullable(false).key(true).build(),
+        new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64).nullable(false).key(false).build()
+    ));
+    CreateTableOptions createOptions = new CreateTableOptions();
+    createOptions.setRangePartitionColumns(ImmutableList.<String>of());
+    createOptions.setNumReplicas(1);
+    syncClient.createTable(testTableName, schema, createOptions);
+
+    KuduTable table = syncClient.openTable(testTableName);
+
+    KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
+    List<KuduScanToken> tokens = tokenBuilder.build();
+    assertEquals(1, tokens.size());
+    KuduScanToken token = tokens.get(0);
+
+    // Drop a column
+    syncClient.alterTable(testTableName, new AlterTableOptions().dropColumn("a"));
+    try {
+      token.intoScanner(syncClient);
+      fail();
+    } catch (IllegalArgumentException e) {
+      assertTrue(e.getMessage().contains("Unknown column"));
+    }
+
+    // Add back the column with the wrong type.
+    syncClient.alterTable(
+        testTableName,
+        new AlterTableOptions().addColumn(
+            new ColumnSchema.ColumnSchemaBuilder("a", Type.STRING).nullable(true).build()));
+    try {
+      token.intoScanner(syncClient);
+      fail();
+    } catch (IllegalStateException e) {
+      assertTrue(e.getMessage().contains(
+          "invalid type INT64 for column 'a' in scan token, expected: STRING"));
+    }
+
+    // Add the column with the wrong nullability.
+    syncClient.alterTable(
+        testTableName,
+        new AlterTableOptions().dropColumn("a")
+                               .addColumn(new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64)
+                                                          .nullable(true).build()));
+    try {
+      token.intoScanner(syncClient);
+      fail();
+    } catch (IllegalStateException e) {
+      assertTrue(e.getMessage().contains(
+          "invalid nullability for column 'a' in scan token, expected: NOT NULL"));
+    }
+
+    // Add the column with the correct type and nullability.
+    syncClient.alterTable(
+        testTableName,
+        new AlterTableOptions().dropColumn("a")
+                               .addColumn(new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64)
+                                                          .nullable(false)
+                                                          .defaultValue(0L).build()));
+    token.intoScanner(syncClient);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/kudu/blob/0cda8c8e/java/kudu-client/src/test/java/org/apache/kudu/util/ClientTestUtil.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/util/ClientTestUtil.java b/java/kudu-client/src/test/java/org/apache/kudu/util/ClientTestUtil.java
index 5476175..055955c 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/util/ClientTestUtil.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/util/ClientTestUtil.java
@@ -18,6 +18,7 @@
 package org.apache.kudu.util;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.stumbleupon.async.Callback;
 import com.stumbleupon.async.Deferred;
@@ -30,8 +31,10 @@ import org.apache.kudu.client.AsyncKuduScanner;
 import org.apache.kudu.client.AsyncKuduSession;
 import org.apache.kudu.client.CreateTableOptions;
 import org.apache.kudu.client.Insert;
+import org.apache.kudu.client.KuduClient;
 import org.apache.kudu.client.KuduException;
 import org.apache.kudu.client.KuduPredicate;
+import org.apache.kudu.client.KuduScanToken;
 import org.apache.kudu.client.KuduScanner;
 import org.apache.kudu.client.KuduTable;
 import org.apache.kudu.client.PartialRow;
@@ -40,6 +43,7 @@ import org.apache.kudu.client.RowResultIterator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -123,6 +127,47 @@ public abstract class ClientTestUtil {
     return countRowsInScan(scanBuilder.build());
   }
 
+  /**
+   * Counts the rows in the provided scan tokens.
+   */
+  public static int countScanTokenRows(List<KuduScanToken> tokens, final String masterAddresses,
+                                       final long operationTimeoutMs)
+      throws IOException, InterruptedException {
+    final AtomicInteger count = new AtomicInteger(0);
+    List<Thread> threads = new ArrayList<>();
+    for (final KuduScanToken token : tokens) {
+      final byte[] serializedToken = token.serialize();
+      Thread thread = new Thread(new Runnable() {
+        @Override
+        public void run() {
+          try (KuduClient contextClient = new KuduClient.KuduClientBuilder(masterAddresses)
+                   .defaultAdminOperationTimeoutMs(operationTimeoutMs)
+                   .build()) {
+            KuduScanner scanner = KuduScanToken.deserializeIntoScanner(serializedToken, contextClient);
+            try {
+              int localCount = 0;
+              while (scanner.hasMoreRows()) {
+                localCount += Iterators.size(scanner.nextRows());
+              }
+              count.addAndGet(localCount);
+            } finally {
+              scanner.close();
+            }
+          } catch (Exception e) {
+            LOG.error("exception in parallel token scanner", e);
+          }
+        }
+      });
+      thread.run();
+      threads.add(thread);
+    }
+
+    for (Thread thread : threads) {
+      thread.join();
+    }
+    return count.get();
+  }
+
   public static List<String> scanTableToStrings(KuduTable table,
                                                 KuduPredicate... predicates) throws Exception
{
     List<String> rowStrings = Lists.newArrayList();


Mime
View raw message