You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by ad...@apache.org on 2017/09/18 23:08:19 UTC
[2/3] kudu git commit: KUDU-1807 (part 2): ban GetTableSchema for
table createdness in clients
KUDU-1807 (part 2): ban GetTableSchema for table createdness in clients
This patch modifies new Java clients to no longer use the create_table_done
field in GetTableSchema RPCs. After the CreateTable RPC returns, the client
will issue a GetTableSchema RPC to build a KuduTable, then enter an
IsCreateTableDone RPC loop until all of the tablets have been created.
There are a couple other changes worth noting:
- openTable no longer waits for table creation before returning. Doing that
now would require at least one IsCreateTableDone RPC, which basically
defeats the purpose of these changes. Moreover, the C++ client doesn't do
this, so I don't see why the Java client should.
- I removed the 'tablesNotServed' logic which had no useful effect because,
as I recently learned, there's retry logic to deal with TABLET_NOT_RUNNING
master errors embedded more deeply in the Java client.
- I removed the master permit acquisition from the "is create table done"
loop, since these loops should no longer produce a thundering herds (as I
suppose they could have due to 'tablesNotServed').
- I modified createTable and alterTable to make waiting optional. The
default behavior is to wait, which was already the case for createTable,
but not for alterTable. I also added an isCreateTableDone method, which is
now somewhat useful as createTable's waiting is optional.
- The IsCreateTableDone and IsAlterTableDone loops now use table IDs to
ensure that they're robust in the face of concurrent operations that may
change table names. This depends on AlterTableResponsePB containing table
IDs, which was only true as of Kudu 0.10. If used against an older server,
the client will throw an exception.
Change-Id: I54fa07dc34a97f1c9da06ec9129d6d4590b7aac6
Reviewed-on: http://gerrit.cloudera.org:8080/8026
Tested-by: Kudu Jenkins
Reviewed-by: Jean-Daniel Cryans <jd...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/c1f78ffb
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/c1f78ffb
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/c1f78ffb
Branch: refs/heads/master
Commit: c1f78ffb59af28c0f453cb8485ed946465f19f43
Parents: 2b95bf1
Author: Adar Dembo <ad...@cloudera.com>
Authored: Sat Sep 9 14:37:56 2017 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Mon Sep 18 23:07:15 2017 +0000
----------------------------------------------------------------------
.../apache/kudu/client/AlterTableOptions.java | 27 +-
.../apache/kudu/client/AlterTableResponse.java | 2 +-
.../org/apache/kudu/client/AsyncKuduClient.java | 618 ++++++++++++-------
.../apache/kudu/client/CreateTableOptions.java | 28 +-
.../apache/kudu/client/CreateTableRequest.java | 5 +-
.../apache/kudu/client/CreateTableResponse.java | 11 +-
.../org/apache/kudu/client/DeadlineTracker.java | 6 +-
.../kudu/client/GetTableSchemaRequest.java | 29 +-
.../kudu/client/GetTableSchemaResponse.java | 15 +-
.../kudu/client/IsAlterTableDoneRequest.java | 18 +-
.../kudu/client/IsCreateTableDoneRequest.java | 36 +-
.../kudu/client/IsCreateTableDoneResponse.java | 45 ++
.../java/org/apache/kudu/client/KuduClient.java | 84 +--
.../org/apache/kudu/client/TestAlterTable.java | 9 -
.../apache/kudu/client/TestDeadlineTracker.java | 18 +-
.../kudu/client/TestFlexiblePartitioning.java | 2 +-
.../org/apache/kudu/client/TestKuduClient.java | 57 +-
.../org/apache/kudu/client/TestKuduTable.java | 103 ++--
18 files changed, 708 insertions(+), 405 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableOptions.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableOptions.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableOptions.java
index 9f1af06..4771c61 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableOptions.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableOptions.java
@@ -36,8 +36,8 @@ import org.apache.kudu.Type;
@InterfaceAudience.Public
@InterfaceStability.Unstable
public class AlterTableOptions {
-
private final AlterTableRequestPB.Builder pb = AlterTableRequestPB.newBuilder();
+ private boolean wait = true;
/**
* Change a table's name.
@@ -357,6 +357,27 @@ public class AlterTableOptions {
}
/**
+ * Whether to wait for the table to be fully altered before this alter
+ * operation is considered to be finished.
+ * <p>
+ * If false, the alter will finish quickly, but a subsequent
+ * {@link KuduClient#openTable(String)} may return a {@link KuduTable} with
+ * an out-of-date schema.
+ * <p>
+ * If true, the alter will take longer, but the very next schema is guaranteed
+ * to be up-to-date.
+ * <p>
+ * If not provided, defaults to true.
+ * <p>
+ * @param wait whether to wait for the table to be fully altered
+ * @return this instance
+ */
+ public AlterTableOptions setWait(boolean wait) {
+ this.wait = wait;
+ return this;
+ }
+
+ /**
* @return {@code true} if the alter table operation includes an add or drop partition operation
*/
@InterfaceAudience.Private
@@ -370,4 +391,8 @@ public class AlterTableOptions {
AlterTableRequestPB.Builder getProtobuf() {
return pb;
}
+
+ boolean shouldWait() {
+ return wait;
+ }
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableResponse.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableResponse.java
index 27b3213..62a3b74 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableResponse.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableResponse.java
@@ -35,7 +35,7 @@ public class AlterTableResponse extends KuduRpcResponse {
}
/**
- * @return the ID of the altered table, or null if the master version is too old
+ * @return the ID of the altered table
*/
public String getTableId() {
return tableId;
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
index aec75e7..f23b342 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
@@ -35,7 +35,6 @@ import java.net.UnknownHostException;
import java.security.cert.CertificateException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
@@ -75,6 +74,7 @@ import org.apache.kudu.Common;
import org.apache.kudu.Schema;
import org.apache.kudu.master.Master;
import org.apache.kudu.master.Master.GetTableLocationsResponsePB;
+import org.apache.kudu.master.Master.TableIdentifierPB;
import org.apache.kudu.util.AsyncUtil;
import org.apache.kudu.util.NetUtil;
import org.apache.kudu.util.Pair;
@@ -172,15 +172,6 @@ public class AsyncKuduClient implements AutoCloseable {
private long lastPropagatedTimestamp = NO_TIMESTAMP;
/**
- * A table is considered not served when we get a TABLET_NOT_RUNNING error from the master
- * after calling GetTableLocations (it means that some tablets aren't ready to serve yet).
- * We cache this information so that concurrent RPCs sent just after creating a table don't
- * all try to hit the master for no good reason.
- */
- private final Set<String> tablesNotServed = Collections.newSetFromMap(new
- ConcurrentHashMap<String, Boolean>());
-
- /**
* Semaphore used to rate-limit master lookups
* Once we have more than this number of concurrent master lookups, we'll
* start to throttle ourselves slightly.
@@ -375,16 +366,64 @@ public class AsyncKuduClient implements AutoCloseable {
"setRangePartitionColumns or addHashPartitions");
}
- CreateTableRequest create = new CreateTableRequest(this.masterTable, name, schema, builder);
+
+ // Send the CreateTable RPC.
+ final CreateTableRequest create = new CreateTableRequest(
+ this.masterTable, name, schema, builder);
create.setTimeoutMillis(defaultAdminOperationTimeoutMs);
- return sendRpcToTablet(create).addCallbackDeferring(
+ Deferred<CreateTableResponse> createTableD = sendRpcToTablet(create);
+
+ // Add a callback that converts the response into a KuduTable.
+ Deferred<KuduTable> kuduTableD = createTableD.addCallbackDeferring(
new Callback<Deferred<KuduTable>, CreateTableResponse>() {
@Override
- public Deferred<KuduTable> call(CreateTableResponse createTableResponse)
- throws Exception {
- return openTable(name);
+ public Deferred<KuduTable> call(CreateTableResponse resp) throws Exception {
+ return getTableSchema(name, resp.getTableId(), create);
}
});
+
+ if (!builder.shouldWait()) {
+ return kuduTableD;
+ }
+
+ // If requested, add a callback that waits until all of the table's tablets
+ // have been created.
+ return kuduTableD.addCallbackDeferring(new Callback<Deferred<KuduTable>, KuduTable>() {
+ @Override
+ public Deferred<KuduTable> call(KuduTable tableResp) throws Exception {
+ TableIdentifierPB.Builder table = TableIdentifierPB.newBuilder()
+ .setTableId(ByteString.copyFromUtf8(tableResp.getTableId()));
+ return getDelayedIsCreateTableDoneDeferred(table, create, tableResp);
+ }
+ });
+ }
+
+ /**
+ * Check whether a previously issued createTable() is done.
+ * @param name table's name
+ * @return a deferred object to track the progress of the isCreateTableDone command
+ */
+ public Deferred<IsCreateTableDoneResponse> isCreateTableDone(String name) {
+ return doIsCreateTableDone(TableIdentifierPB.newBuilder().setTableName(name), null);
+ }
+
+ /**
+ * Check whether a previously issued createTable() is done.
+ * @param table table identifier
+ * @param parent parent RPC (for tracing), if any
+ * @return a deferred object to track the progress of the isCreateTableDone command
+ */
+ private Deferred<IsCreateTableDoneResponse> doIsCreateTableDone(
+ @Nonnull TableIdentifierPB.Builder table,
+ @Nullable KuduRpc<?> parent) {
+ checkIsClosed();
+ IsCreateTableDoneRequest request = new IsCreateTableDoneRequest(
+ this.masterTable, table);
+ request.setTimeoutMillis(defaultAdminOperationTimeoutMs);
+ if (parent != null) {
+ request.setParentRpc(parent);
+ }
+ return sendRpcToTablet(request);
}
/**
@@ -402,31 +441,22 @@ public class AsyncKuduClient implements AutoCloseable {
/**
* Alter a table on the cluster as specified by the builder.
*
- * When the returned deferred completes it only indicates that the master accepted the alter
- * command, use {@link AsyncKuduClient#isAlterTableDone(String)} to know when the alter finishes.
- * @param name the table's name, if this is a table rename then the old table name must be passed
- * @param ato the alter table builder
+ * @param name the table's name (old name if the table is being renamed)
+ * @param ato the alter table options
* @return a deferred object to track the progress of the alter command
*/
public Deferred<AlterTableResponse> alterTable(String name, AlterTableOptions ato) {
checkIsClosed();
- AlterTableRequest alter = new AlterTableRequest(this.masterTable, name, ato);
+ final AlterTableRequest alter = new AlterTableRequest(this.masterTable, name, ato);
alter.setTimeoutMillis(defaultAdminOperationTimeoutMs);
- Deferred<AlterTableResponse> response = sendRpcToTablet(alter);
+ Deferred<AlterTableResponse> responseD = sendRpcToTablet(alter);
if (ato.hasAddDropRangePartitions()) {
// Clear the table locations cache so the new partition is immediately visible.
- return response.addCallback(new Callback<AlterTableResponse, AlterTableResponse>() {
+ responseD = responseD.addCallback(new Callback<AlterTableResponse, AlterTableResponse>() {
@Override
public AlterTableResponse call(AlterTableResponse resp) {
- // If the master is of a recent enough version to return the table ID,
- // we can selectively clear the cache only for the altered table.
- // Otherwise, we clear the caches for all tables.
- if (resp.getTableId() != null) {
- tableLocations.remove(resp.getTableId());
- } else {
- tableLocations.clear();
- }
+ tableLocations.remove(resp.getTableId());
return resp;
}
@@ -449,19 +479,48 @@ public class AsyncKuduClient implements AutoCloseable {
}
});
}
- return response;
+ if (!ato.shouldWait()) {
+ return responseD;
+ }
+
+ // If requested, add a callback that waits until all of the table's tablets
+ // have been altered.
+ return responseD.addCallbackDeferring(
+ new Callback<Deferred<AlterTableResponse>, AlterTableResponse>() {
+ @Override
+ public Deferred<AlterTableResponse> call(AlterTableResponse resp) throws Exception {
+ TableIdentifierPB.Builder table = TableIdentifierPB.newBuilder()
+ .setTableId(ByteString.copyFromUtf8(resp.getTableId()));
+ return getDelayedIsAlterTableDoneDeferred(table, alter, resp);
+ }
+ });
}
/**
- * Helper method that checks and waits until the completion of an alter command.
- * It will block until the alter command is done or the deadline is reached.
- * @param name the table's name, if the table was renamed then that name must be checked against
+ * Check whether a previously issued alterTable() is done.
+ * @param name table name
* @return a deferred object to track the progress of the isAlterTableDone command
*/
public Deferred<IsAlterTableDoneResponse> isAlterTableDone(String name) {
+ return doIsAlterTableDone(TableIdentifierPB.newBuilder().setTableName(name), null);
+ }
+
+ /**
+ * Check whether a previously issued alterTable() is done.
+ * @param table table identifier
+ * @param parent parent RPC (for tracing), if any
+ * @return a deferred object to track the progress of the isAlterTableDone command
+ */
+ private Deferred<IsAlterTableDoneResponse> doIsAlterTableDone(
+ @Nonnull TableIdentifierPB.Builder table,
+ @Nullable KuduRpc<?> parent) {
checkIsClosed();
- IsAlterTableDoneRequest request = new IsAlterTableDoneRequest(this.masterTable, name);
+ IsAlterTableDoneRequest request = new IsAlterTableDoneRequest(
+ this.masterTable, table);
request.setTimeoutMillis(defaultAdminOperationTimeoutMs);
+ if (parent != null) {
+ request.setParentRpc(parent);
+ }
return sendRpcToTablet(request);
}
@@ -476,10 +535,47 @@ public class AsyncKuduClient implements AutoCloseable {
return sendRpcToTablet(rpc);
}
- private Deferred<GetTableSchemaResponse> getTableSchema(String name) {
- GetTableSchemaRequest rpc = new GetTableSchemaRequest(this.masterTable, name);
+ /**
+ * Gets a table's schema either by ID or by name. Note: the name must be
+ * provided, even if the RPC should be sent by ID.
+ * @param name name of table
+ * @param id immutable ID of table
+ * @param parent parent RPC (for tracing), if any
+ * @return a deferred object that yields the schema
+ */
+ private Deferred<KuduTable> getTableSchema(
+ @Nonnull final String tableName,
+ @Nullable String tableId,
+ @Nullable KuduRpc<?> parent) {
+ Preconditions.checkNotNull(tableName);
+
+ // Prefer a lookup by table ID over name, since the former is immutable.
+ GetTableSchemaRequest rpc = new GetTableSchemaRequest(
+ this.masterTable, tableId, tableId != null ? null : tableName);
+
+ if (parent != null) {
+ rpc.setParentRpc(parent);
+ }
rpc.setTimeoutMillis(defaultAdminOperationTimeoutMs);
- return sendRpcToTablet(rpc);
+ return sendRpcToTablet(rpc).addCallback(new Callback<KuduTable, GetTableSchemaResponse>() {
+ @Override
+ public KuduTable call(GetTableSchemaResponse resp) throws Exception {
+ // When opening a table, clear the existing cached non-covered range entries.
+ // This avoids surprises where a new table instance won't be able to see the
+ // current range partitions of a table for up to the ttl.
+ TableLocationsCache cache = tableLocations.get(resp.getTableId());
+ if (cache != null) {
+ cache.clearNonCoveredRangeEntries();
+ }
+
+ LOG.debug("Opened table {}", resp.getTableId());
+ return new KuduTable(AsyncKuduClient.this,
+ tableName,
+ resp.getTableId(),
+ resp.getSchema(),
+ resp.getPartitionSchema());
+ }
+ });
}
/**
@@ -526,9 +622,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
/**
- * Open the table with the given name. If the table was just created, the
- * Deferred will only get called back when all the tablets have been
- * successfully created.
+ * Open the table with the given name.
*
* New range partitions created by other clients will immediately be available
* after opening the table.
@@ -536,105 +630,9 @@ public class AsyncKuduClient implements AutoCloseable {
* @param name table to open
* @return a KuduTable if the table exists, else a MasterErrorException
*/
- public Deferred<KuduTable> openTable(final String name) {
+ public Deferred<KuduTable> openTable(String name) {
checkIsClosed();
-
- // We create an RPC that we're never going to send, and will instead use it to keep track of
- // timeouts and use its Deferred.
- final KuduRpc<KuduTable> fakeRpc = new KuduRpc<KuduTable>(null) {
- @Override
- Message createRequestPB() {
- return null;
- }
-
- @Override
- String serviceName() {
- return null;
- }
-
- @Override
- String method() {
- return "IsCreateTableDone";
- }
-
- @Override
- Pair<KuduTable, Object> deserialize(CallResponse callResponse, String tsUUID)
- throws KuduException {
- return null;
- }
- };
- fakeRpc.setTimeoutMillis(defaultAdminOperationTimeoutMs);
-
- return getTableSchema(name).addCallbackDeferring(new Callback<Deferred<KuduTable>,
- GetTableSchemaResponse>() {
- @Override
- public Deferred<KuduTable> call(GetTableSchemaResponse response) throws Exception {
- KuduTable table = new KuduTable(AsyncKuduClient.this,
- name,
- response.getTableId(),
- response.getSchema(),
- response.getPartitionSchema());
- // We grab the Deferred first because calling callback on the RPC will reset it and we'd
- // return a different, non-triggered Deferred.
- Deferred<KuduTable> d = fakeRpc.getDeferred();
- if (response.isCreateTableDone()) {
- // When opening a table, clear the existing cached non-covered range entries.
- // This avoids surprises where a new table instance won't be able to see the
- // current range partitions of a table for up to the ttl.
- TableLocationsCache cache = tableLocations.get(response.getTableId());
- if (cache != null) {
- cache.clearNonCoveredRangeEntries();
- }
-
- LOG.debug("Opened table {}", name);
- fakeRpc.callback(table);
- } else {
- LOG.debug("Delaying opening table {}, its tablets aren't fully created", name);
- fakeRpc.attempt++;
- delayedIsCreateTableDone(
- table,
- fakeRpc,
- getOpenTableCB(fakeRpc, table),
- getDelayedIsCreateTableDoneErrback(fakeRpc));
- }
- return d;
- }
- });
- }
-
- /**
- * This callback will be repeatedly used when opening a table until it is done being created.
- */
- private Callback<Deferred<KuduTable>, Master.IsCreateTableDoneResponsePB> getOpenTableCB(
- final KuduRpc<KuduTable> rpc, final KuduTable table) {
- return new Callback<Deferred<KuduTable>, Master.IsCreateTableDoneResponsePB>() {
- @Override
- public Deferred<KuduTable> call(
- Master.IsCreateTableDoneResponsePB isCreateTableDoneResponsePB) throws Exception {
- String tableName = table.getName();
- Deferred<KuduTable> d = rpc.getDeferred();
- if (isCreateTableDoneResponsePB.getDone()) {
- // When opening a table, clear the existing cached non-covered range entries.
- TableLocationsCache cache = tableLocations.get(table.getTableId());
- if (cache != null) {
- cache.clearNonCoveredRangeEntries();
- }
- LOG.debug("Table {}'s tablets are now created", tableName);
- rpc.callback(table);
- } else {
- rpc.attempt++;
- LOG.debug("Table {}'s tablets are still not created, further delaying opening it",
- tableName);
-
- delayedIsCreateTableDone(
- table,
- rpc,
- getOpenTableCB(rpc, table),
- getDelayedIsCreateTableDoneErrback(rpc));
- }
- return d;
- }
- };
+ return getTableSchema(name, null, null);
}
/**
@@ -889,11 +887,6 @@ public class AsyncKuduClient implements AutoCloseable {
//
// 2) The tablet is known, but we do not have an active client for the
// leader replica.
- if (tablesNotServed.contains(tableId)) {
- return delayedIsCreateTableDone(request.getTable(), request,
- new RetryRpcCB<R, Master.IsCreateTableDoneResponsePB>(request),
- getDelayedIsCreateTableDoneErrback(request));
- }
Callback<Deferred<R>, Master.GetTableLocationsResponsePB> cb = new RetryRpcCB<>(request);
Callback<Deferred<R>, Exception> eb = new RetryRpcErrback<>(request);
Deferred<Master.GetTableLocationsResponsePB> returnedD =
@@ -967,124 +960,281 @@ public class AsyncKuduClient implements AutoCloseable {
}
/**
- * This errback ensures that if the delayed call to IsCreateTableDone throws an Exception that
+ * Returns an errback ensuring that if the delayed call throws an Exception,
* it will be propagated back to the user.
- * @param request Request to errback if there's a problem with the delayed call.
- * @param <R> Request's return type.
- * @return An errback.
+ * <p>
+ * @param rpc RPC to errback if there's a problem with the delayed call
+ * @param <R> RPC's return type
+ * @return newly created errback
*/
- private <R> Callback<Exception, Exception> getDelayedIsCreateTableDoneErrback(
- final KuduRpc<R> request) {
+ private <R> Callback<Exception, Exception> getDelayedIsTableDoneEB(
+ final KuduRpc<R> rpc) {
return new Callback<Exception, Exception>() {
@Override
public Exception call(Exception e) throws Exception {
// TODO maybe we can retry it?
- request.errback(e);
+ rpc.errback(e);
return e;
}
};
}
/**
- * This method will call IsCreateTableDone on the master after sleeping for
- * getSleepTimeForRpc() based on the provided KuduRpc's number of attempts. Once this is done,
- * the provided callback will be called.
- * @param table the table to lookup
- * @param rpc the original KuduRpc that needs to access the table
- * @param retryCB the callback to call on completion
- * @param errback the errback to call if something goes wrong when calling IsCreateTableDone
- * @return Deferred used to track the provided KuduRpc
+ * Creates an RPC that will never be sent, and will instead be used
+ * exclusively for timeouts.
+ * @param method fake RPC method (shows up in RPC traces)
+ * @param parent parent RPC (for tracing), if any
+ * @param <R> the expected return type of the fake RPC
+ * @return created fake RPC
*/
- private <R> Deferred<R> delayedIsCreateTableDone(
- final KuduTable table,
- final KuduRpc<R> rpc,
- final Callback<Deferred<R>,
- Master.IsCreateTableDoneResponsePB> retryCB,
- final Callback<Exception, Exception> errback) {
+ private <R> KuduRpc<R> buildFakeRpc(
+ @Nonnull final String method,
+ @Nullable final KuduRpc<?> parent) {
+ KuduRpc<R> rpc = new KuduRpc<R>(null) {
+ @Override
+ Message createRequestPB() {
+ return null;
+ }
- final class RetryTimer implements TimerTask {
- public void run(final Timeout timeout) {
- String tableId = table.getTableId();
- final boolean has_permit = acquireMasterLookupPermit();
- if (!has_permit && !tablesNotServed.contains(tableId)) {
- // If we failed to acquire a permit, it's worth checking if someone
- // looked up the tablet we're interested in. Every once in a while
- // this will save us a Master lookup.
- try {
- retryCB.call(null);
- return;
- } catch (Exception e) {
- // we're calling RetryRpcCB which doesn't throw exceptions, ignore
- }
- }
- IsCreateTableDoneRequest isCreateTableDoneRequest =
- new IsCreateTableDoneRequest(masterTable, tableId);
- isCreateTableDoneRequest.setTimeoutMillis(defaultAdminOperationTimeoutMs);
- isCreateTableDoneRequest.setParentRpc(rpc);
- final Deferred<Master.IsCreateTableDoneResponsePB> d =
- sendRpcToTablet(isCreateTableDoneRequest).addCallback(new IsCreateTableDoneCB(tableId));
- if (has_permit) {
- // The errback is needed here to release the lookup permit
- d.addCallbacks(new ReleaseMasterLookupPermit<Master.IsCreateTableDoneResponsePB>(),
- new ReleaseMasterLookupPermit<Exception>());
- }
- d.addCallbacks(retryCB, errback);
+ @Override
+ String serviceName() {
+ return null;
}
- }
- long sleepTime = getSleepTimeForRpc(rpc);
- if (rpc.deadlineTracker.wouldSleepingTimeout(sleepTime)) {
- return tooManyAttemptsOrTimeout(rpc, null);
+ @Override
+ String method() {
+ return method;
+ }
+
+ @Override
+ Pair<R, Object> deserialize(
+ CallResponse callResponse, String tsUUID) throws KuduException {
+ return null;
+ }
+ };
+ rpc.setTimeoutMillis(defaultAdminOperationTimeoutMs);
+ if (parent != null) {
+ rpc.setParentRpc(parent);
}
+ return rpc;
+ }
- newTimeout(new RetryTimer(), sleepTime);
- return rpc.getDeferred();
+ /**
+ * Schedules a IsAlterTableDone RPC. When the response comes in, if the table
+ * is done altering, the RPC's callback chain is triggered with 'resp' as its
+ * value. If not, another IsAlterTableDone RPC is scheduled and the cycle
+ * repeats, until the alter is finished or a timeout is reached.
+ * @param table table identifier
+ * @param parent parent RPC (for tracing), if any
+ * @param resp previous AlterTableResponse, if any
+ * @return Deferred that will become ready when the alter is done
+ */
+ Deferred<AlterTableResponse> getDelayedIsAlterTableDoneDeferred(
+ @Nonnull TableIdentifierPB.Builder table,
+ @Nullable KuduRpc<?> parent,
+ @Nullable AlterTableResponse resp) {
+ // TODO(adar): By scheduling even the first RPC via timer, the sequence of
+ // RPCs is delayed by at least one timer tick, which is unfortunate for the
+ // case where the table is already fully altered.
+ //
+ // Eliminating the delay by sending the first RPC immediately (and
+ // scheduling the rest via timer) would also allow us to replace this "fake"
+ // RPC with a real one.
+ KuduRpc<AlterTableResponse> fakeRpc = buildFakeRpc("IsAlterTableDone", parent);
+
+ // Store the Deferred locally; callback() or errback() on the RPC will
+ // reset it and we'd return a different, non-triggered Deferred.
+ Deferred<AlterTableResponse> fakeRpcD = fakeRpc.getDeferred();
+
+ delayedIsAlterTableDone(
+ table,
+ fakeRpc,
+ getDelayedIsAlterTableDoneCB(fakeRpc, table, resp),
+ getDelayedIsTableDoneEB(fakeRpc));
+ return fakeRpcD;
}
- private final class ReleaseMasterLookupPermit<T> implements Callback<T, T> {
- public T call(final T arg) {
- releaseMasterLookupPermit();
- return arg;
- }
+ /**
+ * Schedules a IsCreateTableDone RPC. When the response comes in, if the table
+ * is done creating, the RPC's callback chain is triggered with 'resp' as its
+ * value. If not, another IsCreateTableDone RPC is scheduled and the cycle
+ * repeats, until the createis finished or a timeout is reached.
+ * @param table table identifier
+ * @param parent parent RPC (for tracing), if any
+ * @param resp previous KuduTable, if any
+ * @return Deferred that will become ready when the create is done
+ */
+ Deferred<KuduTable> getDelayedIsCreateTableDoneDeferred(
+ @Nonnull TableIdentifierPB.Builder table,
+ @Nullable KuduRpc<?> parent,
+ @Nullable KuduTable resp) {
+ // TODO(adar): By scheduling even the first RPC via timer, the sequence of
+ // RPCs is delayed by at least one timer tick, which is unfortunate for the
+ // case where the table is already fully altered.
+ //
+ // Eliminating the delay by sending the first RPC immediately (and
+ // scheduling the rest via timer) would also allow us to replace this "fake"
+ // RPC with a real one.
+ KuduRpc<KuduTable> fakeRpc = buildFakeRpc("IsCreateTableDone", parent);
- public String toString() {
- return "release master lookup permit";
- }
+ // Store the Deferred locally; callback() or errback() on the RPC will
+ // reset it and we'd return a different, non-triggered Deferred.
+ Deferred<KuduTable> fakeRpcD = fakeRpc.getDeferred();
+
+ delayedIsCreateTableDone(
+ table,
+ fakeRpc,
+ getDelayedIsCreateTableDoneCB(fakeRpc, table, resp),
+ getDelayedIsTableDoneEB(fakeRpc));
+ return fakeRpcD;
+ }
+
+ /**
+ * Returns a callback to be called upon completion of an IsAlterTableDone RPC.
+ * If the table is fully altered, triggers the provided rpc's callback chain
+ * with 'alterResp' as its value. Otherwise, sends another IsAlterTableDone
+ * RPC after sleeping.
+ * <p>
+ * @param rpc RPC that initiated this sequence of operations
+ * @param table table identifier
+ * @param alterResp response from an earlier AlterTable RPC, if any
+ * @return callback that will eventually return 'alterResp'
+ */
+ private Callback<Deferred<AlterTableResponse>, IsAlterTableDoneResponse> getDelayedIsAlterTableDoneCB(
+ @Nonnull final KuduRpc<AlterTableResponse> rpc,
+ @Nonnull final TableIdentifierPB.Builder table,
+ @Nullable final AlterTableResponse alterResp) {
+ return new Callback<Deferred<AlterTableResponse>, IsAlterTableDoneResponse>() {
+ @Override
+ public Deferred<AlterTableResponse> call(IsAlterTableDoneResponse resp) throws Exception {
+ // Store the Deferred locally; callback() below will reset it and we'd
+ // return a different, non-triggered Deferred.
+ Deferred<AlterTableResponse> d = rpc.getDeferred();
+ if (resp.isDone()) {
+ rpc.callback(alterResp);
+ } else {
+ rpc.attempt++;
+ delayedIsAlterTableDone(
+ table,
+ rpc,
+ getDelayedIsAlterTableDoneCB(rpc, table, alterResp),
+ getDelayedIsTableDoneEB(rpc));
+ }
+ return d;
+ }
+ };
}
- /** Callback executed when IsCreateTableDone completes. */
- private final class IsCreateTableDoneCB implements Callback<Master.IsCreateTableDoneResponsePB,
- Master.IsCreateTableDoneResponsePB> {
- final String tableName;
+ /**
+ * Returns a callback to be called upon completion of an IsCreateTableDone RPC.
+ * If the table is fully created, triggers the provided rpc's callback chain
+ * with 'tableResp' as its value. Otherwise, sends another IsCreateTableDone
+ * RPC after sleeping.
+ * <p>
+ * @param rpc RPC that initiated this sequence of operations
+ * @param table table identifier
+ * @param tableResp previously constructed KuduTable, if any
+ * @return callback that will eventually return 'tableResp'
+ */
+ private Callback<Deferred<KuduTable>, IsCreateTableDoneResponse> getDelayedIsCreateTableDoneCB(
+ final KuduRpc<KuduTable> rpc,
+ final TableIdentifierPB.Builder table,
+ final KuduTable tableResp) {
+ return new Callback<Deferred<KuduTable>, IsCreateTableDoneResponse>() {
+ @Override
+ public Deferred<KuduTable> call(IsCreateTableDoneResponse resp) throws Exception {
+ // Store the Deferred locally; callback() below will reset it and we'd
+ // return a different, non-triggered Deferred.
+ Deferred<KuduTable> d = rpc.getDeferred();
+ if (resp.isDone()) {
+ rpc.callback(tableResp);
+ } else {
+ rpc.attempt++;
+ delayedIsCreateTableDone(
+ table,
+ rpc,
+ getDelayedIsCreateTableDoneCB(rpc, table, tableResp),
+ getDelayedIsTableDoneEB(rpc));
+ }
+ return d;
+ }
+ };
+ }
- IsCreateTableDoneCB(String tableName) {
- this.tableName = tableName;
+ /**
+ * Schedules a timer to send an IsCreateTableDone RPC to the master after
+ * sleeping for getSleepTimeForRpc() (based on the provided KuduRpc's number
+ * of attempts). When the master responds, the provided callback will be called.
+ * <p>
+ * @param table table identifier
+ * @param rpc original KuduRpc that needs to access the table
+ * @param callback callback to call on completion
+ * @param errback errback to call if something goes wrong
+ */
+ private void delayedIsCreateTableDone(
+ final TableIdentifierPB.Builder table,
+ final KuduRpc<KuduTable> rpc,
+ final Callback<Deferred<KuduTable>, IsCreateTableDoneResponse> callback,
+ final Callback<Exception, Exception> errback) {
+ final class RetryTimer implements TimerTask {
+ public void run(final Timeout timeout) {
+ doIsCreateTableDone(table, rpc).addCallbacks(callback, errback);
+ }
+ }
+ long sleepTimeMillis = getSleepTimeForRpcMillis(rpc);
+ if (rpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTimeMillis)) {
+ tooManyAttemptsOrTimeout(rpc, null);
+ return;
}
+ newTimeout(new RetryTimer(), sleepTimeMillis);
+ }
- public Master.IsCreateTableDoneResponsePB
- call(final Master.IsCreateTableDoneResponsePB response) {
- if (response.getDone()) {
- LOG.debug("Table {} was created", tableName);
- tablesNotServed.remove(tableName);
- } else {
- LOG.debug("Table {} is still being created", tableName);
+ /**
+ * Schedules a timer to send an IsAlterTableDone RPC to the master after
+ * sleeping for getSleepTimeForRpc() (based on the provided KuduRpc's number
+ * of attempts). When the master responds, the provided callback will be called.
+ * <p>
+ * @param table table identifier
+ * @param rpc original KuduRpc that needs to access the table
+ * @param callback callback to call on completion
+ * @param errback errback to call if something goes wrong
+ */
+ private void delayedIsAlterTableDone(
+ final TableIdentifierPB.Builder table,
+ final KuduRpc<AlterTableResponse> rpc,
+ final Callback<Deferred<AlterTableResponse>, IsAlterTableDoneResponse> callback,
+ final Callback<Exception, Exception> errback) {
+ final class RetryTimer implements TimerTask {
+ public void run(final Timeout timeout) {
+ doIsAlterTableDone(table, rpc).addCallbacks(callback, errback);
}
- return response;
+ }
+
+ long sleepTimeMillis = getSleepTimeForRpcMillis(rpc);
+ if (rpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTimeMillis)) {
+ tooManyAttemptsOrTimeout(rpc, null);
+ return;
+ }
+ newTimeout(new RetryTimer(), sleepTimeMillis);
+ }
+
+ private final class ReleaseMasterLookupPermit<T> implements Callback<T, T> {
+ public T call(final T arg) {
+ releaseMasterLookupPermit();
+ return arg;
}
public String toString() {
- return "ask the master if " + tableName + " was created";
+ return "release master lookup permit";
}
}
- private long getSleepTimeForRpc(KuduRpc<?> rpc) {
+ private long getSleepTimeForRpcMillis(KuduRpc<?> rpc) {
int attemptCount = rpc.attempt;
- assert (attemptCount > 0);
if (attemptCount == 0) {
- LOG.warn("Possible bug: attempting to retry an RPC with no attempts. RPC: {}", rpc,
- new Exception("Exception created to collect stack trace"));
- attemptCount = 1;
+ // If this is the first RPC attempt, don't sleep at all.
+ return 0;
}
// Randomized exponential backoff, truncated at 4096ms.
long sleepTime = (long)(Math.pow(2.0, Math.min(attemptCount, 12)) *
@@ -1386,7 +1536,7 @@ public class AsyncKuduClient implements AutoCloseable {
/**
* This methods enable putting RPCs on hold for a period of time determined by
- * {@link #getSleepTimeForRpc(KuduRpc)}. If the RPC is out of time/retries, its errback will
+ * {@link #getSleepTimeForRpcMillis(KuduRpc)}. If the RPC is out of time/retries, its errback will
* be immediately called.
* @param rpc the RPC to retry later
* @param ex the reason why we need to retry
@@ -1414,8 +1564,8 @@ public class AsyncKuduClient implements AutoCloseable {
.callStatus(reasonForRetry)
.build());
- long sleepTime = getSleepTimeForRpc(rpc);
- if (cannotRetryRequest(rpc) || rpc.deadlineTracker.wouldSleepingTimeout(sleepTime)) {
+ long sleepTime = getSleepTimeForRpcMillis(rpc);
+ if (cannotRetryRequest(rpc) || rpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTime)) {
// Don't let it retry.
return tooManyAttemptsOrTimeout(rpc, ex);
}
@@ -1476,14 +1626,8 @@ public class AsyncKuduClient implements AutoCloseable {
public Object call(final GetTableLocationsResponsePB response) {
if (response.hasError()) {
- if (response.getError().getCode() == Master.MasterErrorPB.Code.TABLET_NOT_RUNNING) {
- // Keep a note that the table exists but at least one tablet is not yet running.
- LOG.debug("Table {} has a non-running tablet", table.getName());
- tablesNotServed.add(table.getTableId());
- } else {
- Status status = Status.fromMasterErrorPB(response.getError());
- return new NonRecoverableException(status);
- }
+ Status status = Status.fromMasterErrorPB(response.getError());
+ return new NonRecoverableException(status);
} else {
try {
discoverTablets(table,
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableOptions.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableOptions.java b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableOptions.java
index 7983a57..e20368b 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableOptions.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableOptions.java
@@ -34,9 +34,10 @@ import org.apache.kudu.master.Master;
@InterfaceStability.Evolving
public class CreateTableOptions {
- private Master.CreateTableRequestPB.Builder pb = Master.CreateTableRequestPB.newBuilder();
private final List<PartialRow> splitRows = Lists.newArrayList();
private final List<RangePartition> rangePartitions = Lists.newArrayList();
+ private Master.CreateTableRequestPB.Builder pb = Master.CreateTableRequestPB.newBuilder();
+ private boolean wait = true;
/**
* Add a set of hash partitions to the table.
@@ -186,6 +187,27 @@ public class CreateTableOptions {
return this;
}
+ /**
+ * Whether to wait for the table to be fully created before this create
+ * operation is considered to be finished.
+ * <p>
+ * If false, the create will finish quickly, but subsequent row operations
+ * may take longer as they may need to wait for portions of the table to be
+ * fully created.
+ * <p>
+ * If true, the create will take longer, but the speed of subsequent row
+ * operations will not be impacted.
+ * <p>
+ * If not provided, defaults to true.
+ * <p>
+ * @param wait whether to wait for the table to be fully created
+ * @return this instance
+ */
+ public CreateTableOptions setWait(boolean wait) {
+ this.wait = wait;
+ return this;
+ }
+
Master.CreateTableRequestPB.Builder getBuilder() {
if (!splitRows.isEmpty() || !rangePartitions.isEmpty()) {
pb.setSplitRowsRangeBounds(new Operation.OperationsEncoder()
@@ -202,6 +224,10 @@ public class CreateTableOptions {
}
}
+ boolean shouldWait() {
+ return wait;
+ }
+
final class RangePartition {
private final PartialRow lowerBound;
private final PartialRow upperBound;
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
index 5a63ab1..d5616b5 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
@@ -72,7 +72,10 @@ class CreateTableRequest extends KuduRpc<CreateTableResponse> {
final Master.CreateTableResponsePB.Builder builder = Master.CreateTableResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
CreateTableResponse response =
- new CreateTableResponse(deadlineTracker.getElapsedMillis(), tsUUID);
+ new CreateTableResponse(
+ deadlineTracker.getElapsedMillis(),
+ tsUUID,
+ builder.getTableId().toStringUtf8());
return new Pair<CreateTableResponse, Object>(
response, builder.hasError() ? builder.getError() : null);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableResponse.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableResponse.java
index 6c5ad48..6f4427f 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableResponse.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableResponse.java
@@ -21,11 +21,20 @@ import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
public class CreateTableResponse extends KuduRpcResponse {
+ private final String tableId;
/**
* @param ellapsedMillis Time in milliseconds since RPC creation to now.
*/
- CreateTableResponse(long ellapsedMillis, String tsUUID) {
+ CreateTableResponse(long ellapsedMillis, String tsUUID, String tableId) {
super(ellapsedMillis, tsUUID);
+ this.tableId = tableId;
+ }
+
+ /**
+ * @return the ID of the created table
+ */
+ public String getTableId() {
+ return tableId;
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java b/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
index 53f0098..202a38b 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
@@ -106,14 +106,14 @@ public class DeadlineTracker {
/**
* Utility method to check if sleeping for a specified amount of time would put us past the
* deadline.
- * @param plannedSleepTime number of milliseconds for a planned sleep
+ * @param plannedSleepTimeMillis number of milliseconds for a planned sleep
* @return if the planned sleeps goes past the deadline.
*/
- public boolean wouldSleepingTimeout(long plannedSleepTime) {
+ public boolean wouldSleepingTimeoutMillis(long plannedSleepTimeMillis) {
if (!hasDeadline()) {
return false;
}
- return getMillisBeforeDeadline() - plannedSleepTime <= 0;
+ return getMillisBeforeDeadline() - plannedSleepTimeMillis <= 0;
}
/**
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
index 27578d7..7c8ebc8 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
@@ -21,10 +21,13 @@ import static org.apache.kudu.master.Master.GetTableSchemaRequestPB;
import static org.apache.kudu.master.Master.GetTableSchemaResponsePB;
import static org.apache.kudu.master.Master.TableIdentifierPB;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.kudu.Schema;
+import org.apache.kudu.master.Master.TableIdentifierPB.Builder;
import org.apache.kudu.util.Pair;
/**
@@ -32,21 +35,30 @@ import org.apache.kudu.util.Pair;
*/
@InterfaceAudience.Private
public class GetTableSchemaRequest extends KuduRpc<GetTableSchemaResponse> {
- static final String GET_TABLE_SCHEMA = "GetTableSchema";
+ private final String id;
private final String name;
- GetTableSchemaRequest(KuduTable masterTable, String name) {
+ GetTableSchemaRequest(KuduTable masterTable, String id, String name) {
super(masterTable);
+ Preconditions.checkArgument(id != null ^ name != null,
+ "Only one of table ID or the table name should be provided");
+ this.id = id;
this.name = name;
}
@Override
Message createRequestPB() {
- final GetTableSchemaRequestPB.Builder builder = GetTableSchemaRequestPB.newBuilder();
- TableIdentifierPB tableID =
- TableIdentifierPB.newBuilder().setTableName(name).build();
- builder.setTable(tableID);
+ final GetTableSchemaRequestPB.Builder builder =
+ GetTableSchemaRequestPB.newBuilder();
+ Builder identifierBuilder = TableIdentifierPB.newBuilder();
+ if (id != null) {
+ identifierBuilder.setTableId(ByteString.copyFromUtf8(id));
+ } else {
+ Preconditions.checkNotNull(name);
+ identifierBuilder.setTableName(name);
+ }
+ builder.setTable(identifierBuilder.build());
return builder.build();
}
@@ -57,7 +69,7 @@ public class GetTableSchemaRequest extends KuduRpc<GetTableSchemaResponse> {
@Override
String method() {
- return GET_TABLE_SCHEMA;
+ return "GetTableSchema";
}
@Override
@@ -71,8 +83,7 @@ public class GetTableSchemaRequest extends KuduRpc<GetTableSchemaResponse> {
tsUUID,
schema,
respBuilder.getTableId().toStringUtf8(),
- ProtobufHelper.pbToPartitionSchema(respBuilder.getPartitionSchema(), schema),
- respBuilder.getCreateTableDone());
+ ProtobufHelper.pbToPartitionSchema(respBuilder.getPartitionSchema(), schema));
return new Pair<GetTableSchemaResponse, Object>(
response, respBuilder.hasError() ? respBuilder.getError() : null);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaResponse.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaResponse.java
index 309e0f6..d513c13 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaResponse.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaResponse.java
@@ -26,24 +26,23 @@ public class GetTableSchemaResponse extends KuduRpcResponse {
private final Schema schema;
private final PartitionSchema partitionSchema;
- private final boolean createTableDone;
private final String tableId;
/**
* @param ellapsedMillis Time in milliseconds since RPC creation to now
+ * @param tsUUID the UUID of the tablet server that sent the response
* @param schema the table's schema
+ * @param tableId the UUID of the table in the response
* @param partitionSchema the table's partition schema
*/
GetTableSchemaResponse(long ellapsedMillis,
String tsUUID,
Schema schema,
String tableId,
- PartitionSchema partitionSchema,
- boolean createTableDone) {
+ PartitionSchema partitionSchema) {
super(ellapsedMillis, tsUUID);
this.schema = schema;
this.partitionSchema = partitionSchema;
- this.createTableDone = createTableDone;
this.tableId = tableId;
}
@@ -64,14 +63,6 @@ public class GetTableSchemaResponse extends KuduRpcResponse {
}
/**
- * Tells if the original CreateTable call has completed and the tablets are ready.
- * @return true if the table is created, otherwise false
- */
- public boolean isCreateTableDone() {
- return createTableDone;
- }
-
- /**
* Get the table's unique identifier.
* @return the table's tableId
*/
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
index e6b553f..54f41bc 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
@@ -31,22 +31,18 @@ import org.apache.kudu.util.Pair;
*/
@InterfaceAudience.Private
class IsAlterTableDoneRequest extends KuduRpc<IsAlterTableDoneResponse> {
+ private final TableIdentifierPB.Builder table;
- static final String IS_ALTER_TABLE_DONE = "IsAlterTableDone";
- private final String name;
-
-
- IsAlterTableDoneRequest(KuduTable masterTable, String name) {
+ IsAlterTableDoneRequest(KuduTable masterTable, TableIdentifierPB.Builder table) {
super(masterTable);
- this.name = name;
+ this.table = table;
}
@Override
Message createRequestPB() {
- final IsAlterTableDoneRequestPB.Builder builder = IsAlterTableDoneRequestPB.newBuilder();
- TableIdentifierPB tableID =
- TableIdentifierPB.newBuilder().setTableName(name).build();
- builder.setTable(tableID);
+ final IsAlterTableDoneRequestPB.Builder builder =
+ IsAlterTableDoneRequestPB.newBuilder();
+ builder.setTable(table);
return builder.build();
}
@@ -57,7 +53,7 @@ class IsAlterTableDoneRequest extends KuduRpc<IsAlterTableDoneResponse> {
@Override
String method() {
- return IS_ALTER_TABLE_DONE;
+ return "IsAlterTableDone";
}
@Override
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
index e545b96..6375de4 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
@@ -17,24 +17,23 @@
package org.apache.kudu.client;
-import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import org.apache.yetus.audience.InterfaceAudience;
-
-import org.apache.kudu.master.Master;
+import org.apache.kudu.master.Master.IsCreateTableDoneRequestPB;
+import org.apache.kudu.master.Master.IsCreateTableDoneResponsePB;
+import org.apache.kudu.master.Master.TableIdentifierPB;
import org.apache.kudu.util.Pair;
/**
* Package-private RPC that can only go to a master.
*/
@InterfaceAudience.Private
-class IsCreateTableDoneRequest extends KuduRpc<Master.IsCreateTableDoneResponsePB> {
-
- private final String tableId;
+class IsCreateTableDoneRequest extends KuduRpc<IsCreateTableDoneResponse> {
+ private final TableIdentifierPB.Builder table;
- IsCreateTableDoneRequest(KuduTable table, String tableId) {
- super(table);
- this.tableId = tableId;
+ IsCreateTableDoneRequest(KuduTable masterTable, TableIdentifierPB.Builder table) {
+ super(masterTable);
+ this.table = table;
}
@Override
@@ -48,22 +47,23 @@ class IsCreateTableDoneRequest extends KuduRpc<Master.IsCreateTableDoneResponseP
}
@Override
- Pair<Master.IsCreateTableDoneResponsePB, Object> deserialize(
+ Pair<IsCreateTableDoneResponse, Object> deserialize(
final CallResponse callResponse, String tsUUID) throws KuduException {
- Master.IsCreateTableDoneResponsePB.Builder builder = Master.IsCreateTableDoneResponsePB
- .newBuilder();
+ IsCreateTableDoneResponsePB.Builder builder =
+ IsCreateTableDoneResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
- Master.IsCreateTableDoneResponsePB resp = builder.build();
- return new Pair<Master.IsCreateTableDoneResponsePB, Object>(
+ IsCreateTableDoneResponse resp =
+ new IsCreateTableDoneResponse(deadlineTracker.getElapsedMillis(),
+ tsUUID, builder.getDone());
+ return new Pair<IsCreateTableDoneResponse, Object>(
resp, builder.hasError() ? builder.getError() : null);
}
@Override
Message createRequestPB() {
- final Master.IsCreateTableDoneRequestPB.Builder builder = Master
- .IsCreateTableDoneRequestPB.newBuilder();
- builder.setTable(Master.TableIdentifierPB.newBuilder().setTableId(
- ByteString.copyFromUtf8(tableId)));
+ final IsCreateTableDoneRequestPB.Builder builder =
+ IsCreateTableDoneRequestPB.newBuilder();
+ builder.setTable(table);
return builder.build();
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneResponse.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneResponse.java
new file mode 100644
index 0000000..b306ed9
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneResponse.java
@@ -0,0 +1,45 @@
+// 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 org.apache.yetus.audience.InterfaceAudience;
+import org.apache.yetus.audience.InterfaceStability;
+
+/**
+ * Response to an isCreateTableDone command. Describes whether the table is
+ * still being created.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class IsCreateTableDoneResponse extends KuduRpcResponse {
+
+ private final boolean done;
+
+ IsCreateTableDoneResponse(long elapsedMillis, String tsUUID, boolean done) {
+ super(elapsedMillis, tsUUID);
+ this.done = done;
+ }
+
+ /**
+ * Returns whether the table is done being created.
+ * @return whether table creation is finished
+ */
+ public boolean isDone() {
+ return done;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
index ec4a31f..b4b1d80 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
@@ -29,6 +29,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kudu.Schema;
+import org.apache.kudu.master.Master.TableIdentifierPB;
/**
* A synchronous and thread-safe client for Kudu.
@@ -86,6 +87,29 @@ public class KuduClient implements AutoCloseable {
}
/**
+ * Waits for all of the tablets in a table to be created, or until the
+ * default admin operation timeout is reached.
+ * @param name the table's name
+ * @return true if the table is done being created, or false if the default
+ * admin operation timeout was reached.
+ * @throws KuduException for any error returned by sending RPCs to the master
+ * (e.g. the table does not exist)
+ */
+ public boolean isCreateTableDone(String name) throws KuduException {
+ TableIdentifierPB.Builder table = TableIdentifierPB.newBuilder().setTableName(name);
+ Deferred<KuduTable> d = asyncClient.getDelayedIsCreateTableDoneDeferred(table, null, null);
+ try {
+ joinAndHandleException(d);
+ } catch (KuduException e) {
+ if (e.getStatus().isTimedOut()) {
+ return false;
+ }
+ throw e;
+ }
+ return true;
+ }
+
+ /**
* Delete a table on the cluster with the specified name.
* @param name the table's name
* @return an rpc response object
@@ -98,11 +122,8 @@ public class KuduClient implements AutoCloseable {
/**
* Alter a table on the cluster as specified by the builder.
- *
- * When the method returns it only indicates that the master accepted the alter
- * command, use {@link KuduClient#isAlterTableDone(String)} to know when the alter finishes.
- * @param name the table's name, if this is a table rename then the old table name must be passed
- * @param ato the alter table builder
+ * @param name the table's name (old name if the table is being renamed)
+ * @param ato the alter table options
* @return an rpc response object
* @throws KuduException if anything went wrong
*/
@@ -112,46 +133,26 @@ public class KuduClient implements AutoCloseable {
}
/**
- * Helper method that checks and waits until the completion of an alter command.
- * It will block until the alter command is done or the timeout is reached.
- * @param name Table's name, if the table was renamed then that name must be checked against
- * @return a boolean indicating if the table is done being altered
+ * Waits for all of the tablets in a table to be altered, or until the
+ * default admin operation timeout is reached.
+ * @param name the table's name
+ * @return true if the table is done being altered, or false if the default
+ * admin operation timeout was reached.
* @throws KuduException for any error returned by sending RPCs to the master
+ * (e.g. the table does not exist)
*/
public boolean isAlterTableDone(String name) throws KuduException {
- long totalSleepTime = 0;
- while (totalSleepTime < getDefaultAdminOperationTimeoutMs()) {
- long start = System.currentTimeMillis();
-
- try {
- Deferred<IsAlterTableDoneResponse> d = asyncClient.isAlterTableDone(name);
- IsAlterTableDoneResponse response;
-
- response = d.join(AsyncKuduClient.SLEEP_TIME);
- if (response.isDone()) {
- return true;
- }
-
- // Count time that was slept and see if we need to wait a little more.
- long elapsed = System.currentTimeMillis() - start;
- // Don't oversleep the deadline.
- if (totalSleepTime + AsyncKuduClient.SLEEP_TIME > getDefaultAdminOperationTimeoutMs()) {
- return false;
- }
- // elapsed can be bigger if we slept about 500ms
- if (elapsed <= AsyncKuduClient.SLEEP_TIME) {
- LOG.debug("Alter not done, sleep " + (AsyncKuduClient.SLEEP_TIME - elapsed) +
- " and slept " + totalSleepTime);
- Thread.sleep(AsyncKuduClient.SLEEP_TIME - elapsed);
- totalSleepTime += AsyncKuduClient.SLEEP_TIME;
- } else {
- totalSleepTime += elapsed;
- }
- } catch (Exception ex) {
- throw KuduException.transformException(ex);
+ TableIdentifierPB.Builder table = TableIdentifierPB.newBuilder().setTableName(name);
+ Deferred<AlterTableResponse> d = asyncClient.getDelayedIsAlterTableDoneDeferred(table, null, null);
+ try {
+ joinAndHandleException(d);
+ } catch (KuduException e) {
+ if (e.getStatus().isTimedOut()) {
+ return false;
}
+ throw e;
}
- return false;
+ return true;
}
/**
@@ -197,8 +198,7 @@ public class KuduClient implements AutoCloseable {
}
/**
- * Open the table with the given name. If the table was just created, this
- * method will block until all its tablets have also been created.
+ * Open the table with the given name.
*
* New range partitions created by other clients will immediately be available
* after opening the table.
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/test/java/org/apache/kudu/client/TestAlterTable.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestAlterTable.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestAlterTable.java
index 1499287..57d5c34 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestAlterTable.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestAlterTable.java
@@ -30,8 +30,6 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.junit.Before;
import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.ColumnSchema.CompressionAlgorithm;
@@ -41,8 +39,6 @@ import org.apache.kudu.Type;
import org.apache.kudu.util.Pair;
public class TestAlterTable extends BaseKuduTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(TestKuduClient.class);
private String tableName;
@Before
@@ -116,8 +112,6 @@ public class TestAlterTable extends BaseKuduTest {
.addColumn("addNonNull", Type.INT32, 100)
.addNullableColumn("addNullable", Type.INT32)
.addNullableColumn("addNullableDef", Type.INT32, 200));
- boolean done = syncClient.isAlterTableDone(tableName);
- assertTrue(done);
// Reopen table for the new schema.
table = syncClient.openTable(tableName);
@@ -167,7 +161,6 @@ public class TestAlterTable extends BaseKuduTest {
.changeCompressionAlgorithm(col.getName(), CompressionAlgorithm.SNAPPY)
.changeEncoding(col.getName(), Encoding.RLE)
.changeDefault(col.getName(), 0));
- assertTrue(syncClient.isAlterTableDone(tableName));
// Check for new values.
table = syncClient.openTable(tableName);
@@ -185,8 +178,6 @@ public class TestAlterTable extends BaseKuduTest {
syncClient.alterTable(tableName, new AlterTableOptions()
.renameColumn("c0", "c0Key"));
- boolean done = syncClient.isAlterTableDone(tableName);
- assertTrue(done);
// scanning with the old schema
try {
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
index 00bc983..e5d8adf 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
@@ -50,27 +50,27 @@ public class TestDeadlineTracker {
tracker.setDeadline(500);
assertTrue(tracker.hasDeadline());
assertFalse(tracker.timedOut());
- assertFalse(tracker.wouldSleepingTimeout(499));
- assertTrue(tracker.wouldSleepingTimeout(500));
- assertTrue(tracker.wouldSleepingTimeout(501));
+ assertFalse(tracker.wouldSleepingTimeoutMillis(499));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(500));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(501));
assertEquals(500, tracker.getMillisBeforeDeadline());
// fast forward 200ms
timeToReturn.set(200 * 1000000);
assertTrue(tracker.hasDeadline());
assertFalse(tracker.timedOut());
- assertFalse(tracker.wouldSleepingTimeout(299));
- assertTrue(tracker.wouldSleepingTimeout(300));
- assertTrue(tracker.wouldSleepingTimeout(301));
+ assertFalse(tracker.wouldSleepingTimeoutMillis(299));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(300));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(301));
assertEquals(300, tracker.getMillisBeforeDeadline());
// fast forward another 400ms, so the RPC timed out
timeToReturn.set(600 * 1000000);
assertTrue(tracker.hasDeadline());
assertTrue(tracker.timedOut());
- assertTrue(tracker.wouldSleepingTimeout(299));
- assertTrue(tracker.wouldSleepingTimeout(300));
- assertTrue(tracker.wouldSleepingTimeout(301));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(299));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(300));
+ assertTrue(tracker.wouldSleepingTimeoutMillis(301));
assertEquals(1, tracker.getMillisBeforeDeadline());
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
index c9cd71b..b3ab7ff 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
@@ -45,7 +45,7 @@ public class TestFlexiblePartitioning extends BaseKuduTest {
@Before
public void setTableName() {
- tableName = TestKuduClient.class.getName() + "-" + System.currentTimeMillis();
+ tableName = TestFlexiblePartitioning.class.getName() + "-" + System.currentTimeMillis();
}
private static Schema createSchema() {
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/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 f158c64..39d3b52 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
@@ -39,6 +39,8 @@ 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.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
@@ -688,7 +690,6 @@ public class TestKuduClient extends BaseKuduTest {
// Drop a column
syncClient.alterTable(tableName, new AlterTableOptions().dropColumn("a"));
- assertTrue(syncClient.isAlterTableDone(tableName));
try {
token.intoScanner(syncClient);
fail();
@@ -701,7 +702,6 @@ public class TestKuduClient extends BaseKuduTest {
tableName,
new AlterTableOptions().addColumn(
new ColumnSchema.ColumnSchemaBuilder("a", Type.STRING).nullable(true).build()));
- assertTrue(syncClient.isAlterTableDone(tableName));
try {
token.intoScanner(syncClient);
fail();
@@ -716,7 +716,6 @@ public class TestKuduClient extends BaseKuduTest {
new AlterTableOptions().dropColumn("a")
.addColumn(new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64)
.nullable(true).build()));
- assertTrue(syncClient.isAlterTableDone(tableName));
try {
token.intoScanner(syncClient);
fail();
@@ -732,7 +731,6 @@ public class TestKuduClient extends BaseKuduTest {
.addColumn(new ColumnSchema.ColumnSchemaBuilder("a", Type.INT64)
.nullable(false)
.defaultValue(0L).build()));
- assertTrue(syncClient.isAlterTableDone(tableName));
token.intoScanner(syncClient);
}
@@ -907,7 +905,6 @@ public class TestKuduClient extends BaseKuduTest {
lower.addInt("key", 1);
alter.addRangePartition(lower, upper);
alterClient.alterTable(tableName, alter);
- assertTrue(syncClient.isAlterTableDone(tableName));
}
// Count the number of tablets. The result should still be the same, since
@@ -922,4 +919,54 @@ public class TestKuduClient extends BaseKuduTest {
tokens = tokenBuilder.build();
assertEquals(2, tokens.size());
}
+
+ @Test(timeout = 100000)
+ public void testCreateTableWithConcurrentInsert() throws Exception {
+ KuduTable table = syncClient.createTable(
+ tableName, createManyStringsSchema(), createTableOptions().setWait(false));
+
+ // Insert a row.
+ //
+ // It's very likely that the tablets are still being created, but the client
+ // should transparently retry the insert (and associated master lookup)
+ // until the operation succeeds.
+ Insert insert = table.newInsert();
+ insert.getRow().addString("key", "key_0");
+ insert.getRow().addString("c1", "c1_0");
+ insert.getRow().addString("c2", "c2_0");
+ KuduSession session = syncClient.newSession();
+ OperationResponse resp = session.apply(insert);
+ assertFalse(resp.hasRowError());
+
+ // This won't do anything useful (i.e. if the insert succeeds, we know the
+ // table has been created), but it's here for additional code coverage.
+ assertTrue(syncClient.isCreateTableDone(tableName));
+ }
+
+ @Test(timeout = 100000)
+ public void testCreateTableWithConcurrentAlter() throws Exception {
+ // Kick off an asynchronous table creation.
+ Deferred<KuduTable> d = client.createTable(tableName,
+ createManyStringsSchema(), createTableOptions());
+
+ // Rename the table that's being created to make sure it doesn't interfere
+ // with the "wait for all tablets to be created" behavior of createTable().
+ //
+ // We have to retry this in a loop because we might run before the table
+ // actually exists.
+ while (true) {
+ try {
+ syncClient.alterTable(tableName,
+ new AlterTableOptions().renameTable("foo"));
+ break;
+ } catch (KuduException e) {
+ if (!e.getStatus().isNotFound()) {
+ throw e;
+ }
+ }
+ }
+
+ // If createTable() was disrupted by the alterTable(), this will throw.
+ d.join();
+ }
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c1f78ffb/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduTable.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduTable.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduTable.java
index 0a5ece1..401f8be 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduTable.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduTable.java
@@ -32,16 +32,12 @@ import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
public class TestKuduTable extends BaseKuduTest {
- private static final Logger LOG = LoggerFactory.getLogger(TestKuduTable.class);
-
@Rule
public TestName name = new TestName();
@@ -73,8 +69,7 @@ public class TestKuduTable extends BaseKuduTest {
assertEquals("wrong row", "INT32 key=0, STRING value=NULL", rows.get(0));
// Add a default, checking new rows see the new default and old rows remain the same.
- AlterTableOptions ato = new AlterTableOptions().changeDefault("value", "pizza");
- submitAlterAndCheck(ato, tableName);
+ syncClient.alterTable(tableName, new AlterTableOptions().changeDefault("value", "pizza"));
insertDefaultRow(table, session, 1);
rows = scanTableToStrings(table);
@@ -83,8 +78,7 @@ public class TestKuduTable extends BaseKuduTest {
assertEquals("wrong row", "INT32 key=1, STRING value=pizza", rows.get(1));
// Change the default, checking new rows see the new default and old rows remain the same.
- ato = new AlterTableOptions().changeDefault("value", "taco");
- submitAlterAndCheck(ato, tableName);
+ syncClient.alterTable(tableName, new AlterTableOptions().changeDefault("value", "taco"));
insertDefaultRow(table, session, 2);
@@ -95,8 +89,7 @@ public class TestKuduTable extends BaseKuduTest {
assertEquals("wrong row", "INT32 key=2, STRING value=taco", rows.get(2));
// Remove the default, checking that new rows default to NULL and old rows remain the same.
- ato = new AlterTableOptions().removeDefault("value");
- submitAlterAndCheck(ato, tableName);
+ syncClient.alterTable(tableName, new AlterTableOptions().removeDefault("value"));
insertDefaultRow(table, session, 3);
@@ -118,10 +111,10 @@ public class TestKuduTable extends BaseKuduTest {
ColumnSchema.CompressionAlgorithm.NO_COMPRESSION,
table.getSchema().getColumn("value").getCompressionAlgorithm());
- ato = new AlterTableOptions().changeDesiredBlockSize("value", 8192)
+ syncClient.alterTable(tableName, new AlterTableOptions()
+ .changeDesiredBlockSize("value", 8192)
.changeEncoding("value", ColumnSchema.Encoding.DICT_ENCODING)
- .changeCompressionAlgorithm("value", ColumnSchema.CompressionAlgorithm.SNAPPY);
- submitAlterAndCheck(ato, tableName);
+ .changeCompressionAlgorithm("value", ColumnSchema.CompressionAlgorithm.SNAPPY));
KuduTable reopenedTable = syncClient.openTable(tableName);
assertEquals("wrong block size post alter",
@@ -151,39 +144,35 @@ public class TestKuduTable extends BaseKuduTest {
try {
// Add a col.
- AlterTableOptions ato = new AlterTableOptions().addColumn("testaddint", Type.INT32, 4);
- submitAlterAndCheck(ato, tableName);
+ syncClient.alterTable(tableName,
+ new AlterTableOptions().addColumn("testaddint", Type.INT32, 4));
// Rename that col.
- ato = new AlterTableOptions().renameColumn("testaddint", "newtestaddint");
- submitAlterAndCheck(ato, tableName);
+ syncClient.alterTable(tableName,
+ new AlterTableOptions().renameColumn("testaddint", "newtestaddint"));
// Delete it.
- ato = new AlterTableOptions().dropColumn("newtestaddint");
- submitAlterAndCheck(ato, tableName);
+ syncClient.alterTable(tableName, new AlterTableOptions().dropColumn("newtestaddint"));
String newTableName = tableName +"new";
// Rename our table.
- ato = new AlterTableOptions().renameTable(newTableName);
- submitAlterAndCheck(ato, tableName, newTableName);
+ syncClient.alterTable(tableName, new AlterTableOptions().renameTable(newTableName));
// Rename it back.
- ato = new AlterTableOptions().renameTable(tableName);
- submitAlterAndCheck(ato, newTableName, tableName);
+ syncClient.alterTable(newTableName, new AlterTableOptions().renameTable(tableName));
// Add 3 columns, where one has default value, nullable and Timestamp with default value
- ato = new AlterTableOptions()
+ syncClient.alterTable(tableName, new AlterTableOptions()
.addColumn("testaddmulticolnotnull", Type.INT32, 4)
.addNullableColumn("testaddmulticolnull", Type.STRING)
.addColumn("testaddmulticolTimestampcol", Type.UNIXTIME_MICROS,
- (System.currentTimeMillis() * 1000));
- submitAlterAndCheck(ato, tableName);
+ (System.currentTimeMillis() * 1000)));
// Try altering a table that doesn't exist.
String nonExistingTableName = "table_does_not_exist";
try {
- syncClient.alterTable(nonExistingTableName, ato);
+ syncClient.alterTable(nonExistingTableName, new AlterTableOptions());
fail("Shouldn't be able to alter a table that doesn't exist");
} catch (KuduException ex) {
assertTrue(ex.getStatus().isNotFound());
@@ -205,23 +194,6 @@ public class TestKuduTable extends BaseKuduTest {
}
/**
- * Helper method to submit an Alter and wait for it to happen, using the default table name to
- * check.
- */
- private void submitAlterAndCheck(AlterTableOptions ato, String tableToAlter)
- throws Exception {
- submitAlterAndCheck(ato, tableToAlter, tableToAlter);
- }
-
- private void submitAlterAndCheck(AlterTableOptions ato,
- String tableToAlter, String tableToCheck) throws
- Exception {
- AlterTableResponse alterResponse = syncClient.alterTable(tableToAlter, ato);
- boolean done = syncClient.isAlterTableDone(tableToCheck);
- assertTrue(done);
- }
-
- /**
* Test creating tables of different sizes and see that we get the correct number of tablets back
* @throws Exception
*/
@@ -602,4 +574,47 @@ public class TestKuduTable extends BaseKuduTest {
}
return table;
}
+
+ @Test(timeout = 100000)
+ public void testAlterNoWait() throws Exception {
+ String tableName = name.getMethodName() + System.currentTimeMillis();
+ createTable(tableName, basicSchema, getBasicCreateTableOptions());
+
+ String oldName = "column2_i";
+ for (int i = 0; i < 10; i++) {
+ String newName = String.format("foo%d", i);
+ syncClient.alterTable(tableName, new AlterTableOptions()
+ .renameColumn(oldName, newName)
+ .setWait(false));
+
+ // Reload the schema and test for the existence of 'oldName'. Since we
+ // didn't wait for alter to finish, we should be able to find it. However,
+ // this is timing dependent: if the alter finishes before we reload the
+ // schema, loop and try again.
+ try {
+ syncClient.openTable(tableName).getSchema().getColumn(newName);
+ LOG.info("Alter finished too quickly (new column name {} is already " +
+ "visible), trying again", oldName);
+ oldName = newName;
+ continue;
+ } catch (IllegalArgumentException e) {}
+
+ try {
+ syncClient.openTable(tableName).getSchema().getColumn(newName);
+ fail(String.format("New column name %s should have been visible", newName));
+ } catch (IllegalArgumentException e) {}
+
+ // After waiting for the alter to finish and reloading the schema,
+ // the new column name should be visible.
+ syncClient.isAlterTableDone(tableName);
+ try {
+ syncClient.openTable(tableName).getSchema().getColumn(oldName);
+ fail(String.format("Old column name %s should have been visible", oldName));
+ } catch (IllegalArgumentException e) {}
+ syncClient.openTable(tableName).getSchema().getColumn(newName);
+ LOG.info("Test passed on attempt {}", i + 1);
+ return;
+ }
+ fail("Could not run test even after multiple attempts");
+ }
}