You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2018/07/03 02:59:27 UTC
[4/4] kudu git commit: Add scan request timeout to Kudu scan token API
Add scan request timeout to Kudu scan token API
Includes Java and C++ tests.
Change-Id: I151cd1da27419361f507e9ad29276c2b50153077
Reviewed-on: http://gerrit.cloudera.org:8080/10653
Reviewed-by: Dan Burkert <da...@apache.org>
Tested-by: Mike Percy <mp...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/1ae050e4
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/1ae050e4
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/1ae050e4
Branch: refs/heads/master
Commit: 1ae050e4d57bc333de28bcbc62e072e8bafd04b3
Parents: 0cda8c8
Author: Mike Percy <mp...@apache.org>
Authored: Thu Jun 7 23:14:20 2018 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Tue Jul 3 02:58:35 2018 +0000
----------------------------------------------------------------------
.../apache/kudu/client/AsyncKuduScanner.java | 8 +++++++
.../org/apache/kudu/client/KuduScanToken.java | 5 ++++
.../org/apache/kudu/client/KuduScanner.java | 8 +++++++
.../org/apache/kudu/client/TestScanToken.java | 20 ++++++++++++++++
.../org/apache/kudu/util/ClientTestUtil.java | 23 ++++++++++++++++++
src/kudu/client/client.proto | 5 ++++
src/kudu/client/scan_token-internal.cc | 5 ++++
src/kudu/client/scan_token-test.cc | 25 ++++++++++++++++++++
8 files changed, 99 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
index 0d13fa6..dd61bf4 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
@@ -379,6 +379,14 @@ public final class AsyncKuduScanner {
}
/**
+ * Returns the scan request timeout for this scanner.
+ * @return the scan request timeout, in milliseconds
+ */
+ public long getScanRequestTimeout() {
+ return scanRequestTimeout;
+ }
+
+ /**
* Returns the projection schema of this scanner. If specific columns were
* not specified during scanner creation, the table schema is returned.
* @return the projection schema for this scanner
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
index 615deed..7e0e2d9 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
@@ -257,6 +257,10 @@ public class KuduScanToken implements Comparable<KuduScanToken> {
builder.batchSizeBytes(message.getBatchSizeBytes());
}
+ if (message.hasScanRequestTimeoutMs()) {
+ builder.scanRequestTimeout(message.getScanRequestTimeoutMs());
+ }
+
return builder.build();
}
@@ -367,6 +371,7 @@ public class KuduScanToken implements Comparable<KuduScanToken> {
proto.setCacheBlocks(cacheBlocks);
proto.setFaultTolerant(isFaultTolerant);
proto.setBatchSizeBytes(batchSizeBytes);
+ proto.setScanRequestTimeoutMs(scanRequestTimeout);
try {
PartitionPruner pruner = PartitionPruner.create(this);
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
index 7437d1e..13602a5 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
@@ -129,6 +129,14 @@ public class KuduScanner {
}
/**
+ * Returns the current value of the scanner's scan request timeout.
+ * @return the timeout value, in milliseconds
+ */
+ public long getScanRequestTimeout() {
+ return asyncScanner.getScanRequestTimeout();
+ }
+
+ /**
* A Builder class to build {@link KuduScanner}.
* Use {@link KuduClient#newScannerBuilder} in order to get a builder instance.
*/
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/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
index 1761cf7..cf8ba44 100644
--- 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
@@ -24,10 +24,13 @@ import org.apache.kudu.Type;
import org.junit.Before;
import org.junit.Test;
+import java.io.IOException;
import java.util.List;
import static org.apache.kudu.util.ClientTestUtil.countScanTokenRows;
+import static org.apache.kudu.util.ClientTestUtil.createDefaultTable;
import static org.apache.kudu.util.ClientTestUtil.createManyStringsSchema;
+import static org.apache.kudu.util.ClientTestUtil.loadDefaultTable;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -232,4 +235,21 @@ public class TestScanToken extends BaseKuduTest {
token.intoScanner(syncClient);
}
+ /** Test that scanRequestTimeout makes it from the scan token to the underlying Scanner class. */
+ @Test
+ public void testScanRequestTimeout() throws IOException {
+ final int NUM_ROWS_DESIRED = 100;
+ final int SCAN_REQUEST_TIMEOUT_MS = 20;
+ KuduTable table = createDefaultTable(syncClient, testTableName);
+ loadDefaultTable(syncClient, testTableName, NUM_ROWS_DESIRED);
+ KuduScanToken.KuduScanTokenBuilder builder =
+ new KuduScanToken.KuduScanTokenBuilder(client, table);
+ builder.scanRequestTimeout(SCAN_REQUEST_TIMEOUT_MS);
+ List<KuduScanToken> tokens = builder.build();
+ for (KuduScanToken token : tokens) {
+ byte[] serialized = token.serialize();
+ KuduScanner scanner = KuduScanToken.deserializeIntoScanner(serialized, syncClient);
+ assertEquals(SCAN_REQUEST_TIMEOUT_MS, scanner.getScanRequestTimeout());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/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 055955c..d795b26 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
@@ -36,6 +36,7 @@ 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.KuduSession;
import org.apache.kudu.client.KuduTable;
import org.apache.kudu.client.PartialRow;
import org.apache.kudu.client.RowResult;
@@ -262,6 +263,28 @@ public abstract class ClientTestUtil {
return option;
}
+ /**
+ * A generic helper function to create a table with default test options.
+ */
+ public static KuduTable createDefaultTable(KuduClient client, String tableName) throws KuduException {
+ return client.createTable(tableName, getBasicSchema(), getBasicCreateTableOptions());
+ }
+
+ /**
+ * Load a table of default schema with the specified number of records, in ascending key order.
+ */
+ public static void loadDefaultTable(KuduClient client, String tableName, int numRows)
+ throws KuduException {
+ KuduTable table = client.openTable(tableName);
+ KuduSession session = client.newSession();
+ for (int i = 0; i < numRows; i++) {
+ Insert insert = createBasicSchemaInsert(table, i);
+ session.apply(insert);
+ }
+ session.flush();
+ session.close();
+ }
+
public static Insert createBasicSchemaInsert(KuduTable table, int key) {
Insert insert = table.newInsert();
PartialRow row = insert.getRow();
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/src/kudu/client/client.proto
----------------------------------------------------------------------
diff --git a/src/kudu/client/client.proto b/src/kudu/client/client.proto
index eff26e9..140c430 100644
--- a/src/kudu/client/client.proto
+++ b/src/kudu/client/client.proto
@@ -101,6 +101,11 @@ message ScanTokenPB {
// The replica selection policy for the scan request.
// See common.proto for further information about replica selections.
optional ReplicaSelection replica_selection = 16 [default = LEADER_ONLY];
+
+ // The maximum length of time, in milliseconds, that each scan request to a
+ // server can take. If not set, the default value is controlled by the client
+ // scanner implementation.
+ optional int64 scan_request_timeout_ms = 17;
}
// All of the data necessary to authenticate to a cluster from a client with
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/src/kudu/client/scan_token-internal.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/scan_token-internal.cc b/src/kudu/client/scan_token-internal.cc
index 21fc786..2ce965f 100644
--- a/src/kudu/client/scan_token-internal.cc
+++ b/src/kudu/client/scan_token-internal.cc
@@ -201,6 +201,10 @@ Status KuduScanToken::Data::PBIntoScanner(KuduClient* client,
RETURN_NOT_OK(scan_builder->SetBatchSizeBytes(message.batch_size_bytes()));
}
+ if (message.has_scan_request_timeout_ms()) {
+ scan_builder->SetTimeoutMillis(message.scan_request_timeout_ms());
+ }
+
*scanner = scan_builder.release();
return Status::OK();
}
@@ -272,6 +276,7 @@ Status KuduScanTokenBuilder::Data::Build(vector<KuduScanToken*>* tokens) {
pb.set_cache_blocks(configuration_.spec().cache_blocks());
pb.set_fault_tolerant(configuration_.is_fault_tolerant());
pb.set_propagated_timestamp(client->GetLatestObservedTimestamp());
+ pb.set_scan_request_timeout_ms(configuration_.timeout().ToMilliseconds());
if (configuration_.has_batch_size_bytes()) {
pb.set_batch_size_bytes(configuration_.batch_size_bytes());
http://git-wip-us.apache.org/repos/asf/kudu/blob/1ae050e4/src/kudu/client/scan_token-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/scan_token-test.cc b/src/kudu/client/scan_token-test.cc
index bc6bc07..6cde3fd 100644
--- a/src/kudu/client/scan_token-test.cc
+++ b/src/kudu/client/scan_token-test.cc
@@ -30,6 +30,7 @@
#include "kudu/client/client.h"
#include "kudu/client/client.pb.h"
#include "kudu/client/scan_batch.h"
+#include "kudu/client/scan_configuration.h"
#include "kudu/client/scan_predicate.h"
#include "kudu/client/scanner-internal.h"
#include "kudu/client/schema.h"
@@ -45,6 +46,7 @@
#include "kudu/tserver/mini_tablet_server.h"
#include "kudu/tserver/tablet_server.h"
#include "kudu/tserver/tserver.pb.h"
+#include "kudu/util/monotime.h"
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
@@ -296,6 +298,29 @@ TEST_F(ScanTokenTest, TestScanTokens) {
ASSERT_EQ(60, CountRows(tokens));
NO_FATALS(VerifyTabletInfo(tokens));
}
+
+ { // Scan timeout.
+ vector<KuduScanToken*> tokens;
+ ElementDeleter deleter(&tokens);
+ KuduScanTokenBuilder builder(table.get());
+ const int64_t kTimeoutMillis = 300001; // 5 minutes + 1 ms.
+ ASSERT_OK(builder.SetTimeoutMillis(kTimeoutMillis));
+ ASSERT_OK(builder.Build(&tokens));
+
+ for (const auto& token : tokens) {
+ string buf;
+ ASSERT_OK(token->Serialize(&buf));
+ KuduScanner* scanner_raw;
+ ASSERT_OK(KuduScanToken::DeserializeIntoScanner(client_.get(), buf, &scanner_raw));
+ unique_ptr<KuduScanner> scanner(scanner_raw); // Caller gets ownership of the scanner.
+ // Ensure the timeout configuration gets carried through the serialization process.
+ ASSERT_EQ(kTimeoutMillis, scanner->data_->configuration().timeout().ToMilliseconds());
+ }
+
+ ASSERT_GE(8, tokens.size());
+ ASSERT_EQ(200, CountRows(tokens));
+ NO_FATALS(VerifyTabletInfo(tokens));
+ }
}
TEST_F(ScanTokenTest, TestScanTokensWithNonCoveringRange) {