You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by jd...@apache.org on 2016/10/27 23:09:06 UTC
[1/5] kudu git commit: TODO cleanup in TransactionManager
Repository: kudu
Updated Branches:
refs/heads/master 5a46e6076 -> b333a36f1
TODO cleanup in TransactionManager
This removes a couple of TODOs that were now wrong because we now
have exactly-once retry semantics.
The first was about not having to perform the actual waiting for commit
wait on followers. With the new retry semantics this could easily mean
that upon retrying a request on a follower a client would observe the
result of the operation before it safe to do so.
The second was about not peforming the waiting for commit wait if the
RPCs deadline was expired. A client can now have a total deadline that
is higher than a specific RPC's deadline and has exactly once semantics
meaning even if the specific RPC's deadline is expired the client might
still retry within the context of the global deadline. In that case
we wouldn't want to have the client observe the operation before it
was supposed to.
Change-Id: Ib41a5a65dee29d26f0e7c72b4f2d30f1267c8cf2
Reviewed-on: http://gerrit.cloudera.org:8080/4826
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/f2a39c62
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/f2a39c62
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/f2a39c62
Branch: refs/heads/master
Commit: f2a39c624abcd605d64d6301c3d5e307557c4127
Parents: 5a46e60
Author: David Alves <dr...@apache.org>
Authored: Wed Oct 12 16:15:02 2016 -0700
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Wed Oct 26 02:32:40 2016 +0000
----------------------------------------------------------------------
src/kudu/tablet/transactions/transaction_driver.cc | 8 ++------
1 file changed, 2 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2a39c62/src/kudu/tablet/transactions/transaction_driver.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/transactions/transaction_driver.cc b/src/kudu/tablet/transactions/transaction_driver.cc
index 60af908..3a10c16 100644
--- a/src/kudu/tablet/transactions/transaction_driver.cc
+++ b/src/kudu/tablet/transactions/transaction_driver.cc
@@ -479,7 +479,6 @@ void TransactionDriver::ApplyTask() {
// until now.earliest > prepare_latest. Only after this are the locks
// released.
if (mutable_state()->external_consistency_mode() == COMMIT_WAIT) {
- // TODO: only do this on the leader side
TRACE("APPLY: Commit Wait.");
// If we can't commit wait and have already applied we might have consistency
// issues if we still reply to the client that the operation was a success.
@@ -505,11 +504,8 @@ void TransactionDriver::SetResponseTimestamp(TransactionState* transaction_state
Status TransactionDriver::CommitWait() {
MonoTime before = MonoTime::Now();
DCHECK(mutable_state()->external_consistency_mode() == COMMIT_WAIT);
- // TODO: we could plumb the RPC deadline in here, and not bother commit-waiting
- // if the deadline is already expired.
- RETURN_NOT_OK(
- mutable_state()->tablet_peer()->clock()->WaitUntilAfter(mutable_state()->timestamp(),
- MonoTime::Max()));
+ RETURN_NOT_OK(mutable_state()->tablet_peer()->clock()->WaitUntilAfter(
+ mutable_state()->timestamp(), MonoTime::Max()));
mutable_state()->mutable_metrics()->commit_wait_duration_usec =
(MonoTime::Now() - before).ToMicroseconds();
return Status::OK();
[4/5] kudu git commit: KUDU-1715. Add a way to set ReplicaSelection
to the java client
Posted by jd...@apache.org.
KUDU-1715. Add a way to set ReplicaSelection to the java client
This patch adds a ReplicaSelection option for all the RPCs, which always defaults
to LEADER_ONLY, as well as a way to change it for scanners.
Change-Id: I3bb08c9c78271a0065c8aa8fb9b0f3301f84e828
Reviewed-on: http://gerrit.cloudera.org:8080/4837
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/a7fd4fc8
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/a7fd4fc8
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/a7fd4fc8
Branch: refs/heads/master
Commit: a7fd4fc85c0878570a5ab51a8e28275fd4459ff5
Parents: a326fc9
Author: Jean-Daniel Cryans <jd...@apache.org>
Authored: Tue Oct 25 09:51:42 2016 -0700
Committer: Jean-Daniel Cryans <jd...@apache.org>
Committed: Thu Oct 27 18:40:35 2016 +0000
----------------------------------------------------------------------
.../kudu/client/AbstractKuduScannerBuilder.java | 12 +++++++
.../org/apache/kudu/client/AsyncKuduClient.java | 17 +++++----
.../apache/kudu/client/AsyncKuduScanner.java | 21 +++++++-----
.../java/org/apache/kudu/client/KuduRpc.java | 4 +++
.../org/apache/kudu/client/KuduScanner.java | 2 +-
.../org/apache/kudu/client/RemoteTablet.java | 17 +++++++++
.../apache/kudu/client/ReplicaSelection.java | 36 ++++++++++++++++++++
.../apache/kudu/client/TestRemoteTablet.java | 8 +++++
.../kudu/client/TestScannerMultiTablet.java | 15 ++++++++
9 files changed, 115 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
index c23e07e..f58ae41 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
@@ -54,6 +54,7 @@ public abstract class AbstractKuduScannerBuilder
List<String> projectedColumnNames = null;
List<Integer> projectedColumnIndexes = null;
long scanRequestTimeout;
+ ReplicaSelection replicaSelection = ReplicaSelection.LEADER_ONLY;
AbstractKuduScannerBuilder(AsyncKuduClient client, KuduTable table) {
this.client = client;
@@ -313,6 +314,17 @@ public abstract class AbstractKuduScannerBuilder
}
/**
+ * Sets the replica selection mechanism for this scanner. The default is to read from the
+ * currently known leader.
+ * @param replicaSelection replication selection mechanism to use
+ * @return this instance
+ */
+ public S replicaSelection(ReplicaSelection replicaSelection) {
+ this.replicaSelection = replicaSelection;
+ return (S) this;
+ }
+
+ /**
* Set an encoded (inclusive) start partition key for the scan.
*
* @param partitionKey the encoded partition key
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/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 b36737b..35cd9d6 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
@@ -586,8 +586,9 @@ public class AsyncKuduClient implements AutoCloseable {
Deferred<AsyncKuduScanner.Response> scanNextRows(final AsyncKuduScanner scanner) {
RemoteTablet tablet = scanner.currentTablet();
assert (tablet != null);
- TabletClient client = connectionCache.getClient(tablet.getLeaderUUID());
KuduRpc<AsyncKuduScanner.Response> nextRequest = scanner.getNextRowsRequest();
+ TabletClient client =
+ connectionCache.getClient(tablet.getReplicaSelectedUUID(nextRequest.getReplicaSelection()));
Deferred<AsyncKuduScanner.Response> d = nextRequest.getDeferred();
// Important to increment the attempts before the next if statement since
// getSleepTimeForRpc() relies on it if the client is null or dead.
@@ -615,7 +616,9 @@ public class AsyncKuduClient implements AutoCloseable {
return Deferred.fromResult(null);
}
- final TabletClient client = connectionCache.getClient(tablet.getLeaderUUID());
+ final KuduRpc<AsyncKuduScanner.Response> closeRequest = scanner.getCloseRequest();
+ final TabletClient client = connectionCache.getClient(
+ tablet.getReplicaSelectedUUID(closeRequest.getReplicaSelection()));
if (client == null || !client.isAlive()) {
// Oops, we couldn't find a tablet server that hosts this tablet. Our
// cache was probably invalidated while the client was scanning. So
@@ -623,10 +626,10 @@ public class AsyncKuduClient implements AutoCloseable {
LOG.warn("Cannot close {} properly, no connection open for {}", scanner, tablet);
return Deferred.fromResult(null);
}
- final KuduRpc<AsyncKuduScanner.Response> close_request = scanner.getCloseRequest();
- final Deferred<AsyncKuduScanner.Response> d = close_request.getDeferred();
- close_request.attempt++;
- client.sendRpc(close_request);
+
+ final Deferred<AsyncKuduScanner.Response> d = closeRequest.getDeferred();
+ closeRequest.attempt++;
+ client.sendRpc(closeRequest);
return d;
}
@@ -670,7 +673,7 @@ public class AsyncKuduClient implements AutoCloseable {
// If we found a tablet, we'll try to find the TS to talk to.
if (entry != null) {
RemoteTablet tablet = entry.getTablet();
- String uuid = tablet.getLeaderUUID();
+ String uuid = tablet.getReplicaSelectedUUID(request.getReplicaSelection());
if (uuid != null) {
Deferred<R> d = request.getDeferred();
request.setTablet(tablet);
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/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 c61fbdb..b51394a 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
@@ -178,6 +178,8 @@ public final class AsyncKuduScanner {
private final long htTimestamp;
+ private final ReplicaSelection replicaSelection;
+
/////////////////////
// Runtime variables.
/////////////////////
@@ -212,8 +214,6 @@ public final class AsyncKuduScanner {
private Deferred<RowResultIterator> prefetcherDeferred;
- private boolean inFirstTablet = true;
-
final long scanRequestTimeout;
AsyncKuduScanner(AsyncKuduClient client, KuduTable table, List<String> projectedNames,
@@ -222,7 +222,8 @@ public final class AsyncKuduScanner {
Map<String, KuduPredicate> predicates, long limit,
boolean cacheBlocks, boolean prefetching,
byte[] startPrimaryKey, byte[] endPrimaryKey,
- long htTimestamp, int batchSizeBytes, PartitionPruner pruner) {
+ long htTimestamp, int batchSizeBytes, PartitionPruner pruner,
+ ReplicaSelection replicaSelection) {
checkArgument(batchSizeBytes > 0, "Need a strictly positive number of bytes, " +
"got %s", batchSizeBytes);
checkArgument(limit > 0, "Need a strictly positive number for the limit, " +
@@ -280,6 +281,8 @@ public final class AsyncKuduScanner {
this.hasMore = false;
this.closed = true;
}
+
+ this.replicaSelection = replicaSelection;
}
/**
@@ -586,11 +589,6 @@ public final class AsyncKuduScanner {
*/
KuduRpc<Response> getOpenRequest() {
checkScanningNotStarted();
- // This is the only point where we know we haven't started scanning and where the scanner
- // should be fully configured
- if (this.inFirstTablet) {
- this.inFirstTablet = false;
- }
return new ScanRequest(table, State.OPENING);
}
@@ -684,6 +682,11 @@ public final class AsyncKuduScanner {
}
}
+ @Override
+ ReplicaSelection getReplicaSelection() {
+ return replicaSelection;
+ }
+
/** Serializes this request. */
ChannelBuffer serialize(Message header) {
final ScanRequestPB.Builder builder = ScanRequestPB.newBuilder();
@@ -818,7 +821,7 @@ public final class AsyncKuduScanner {
client, table, projectedColumnNames, projectedColumnIndexes, readMode, orderMode,
scanRequestTimeout, predicates, limit, cacheBlocks,
prefetching, lowerBoundPrimaryKey, upperBoundPrimaryKey,
- htTimestamp, batchSizeBytes, PartitionPruner.create(this));
+ htTimestamp, batchSizeBytes, PartitionPruner.create(this), replicaSelection);
}
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
index 02b547e..8b5c0df 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
@@ -262,6 +262,10 @@ public abstract class KuduRpc<R> {
return sequenceId;
}
+ ReplicaSelection getReplicaSelection() {
+ return ReplicaSelection.LEADER_ONLY;
+ }
+
void setSequenceId(long sequenceId) {
assert (this.sequenceId == RequestTracker.NO_SEQ_NO);
this.sequenceId = sequenceId;
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/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 4a4bcc1..1cccd10 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
@@ -142,7 +142,7 @@ public class KuduScanner {
client, table, projectedColumnNames, projectedColumnIndexes, readMode, orderMode,
scanRequestTimeout, predicates, limit, cacheBlocks,
prefetching, lowerBoundPrimaryKey, upperBoundPrimaryKey,
- htTimestamp, batchSizeBytes, PartitionPruner.create(this)));
+ htTimestamp, batchSizeBytes, PartitionPruner.create(this), replicaSelection));
}
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
index 28be0fe..d98f21d 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
@@ -165,6 +165,23 @@ class RemoteTablet implements Comparable<RemoteTablet> {
}
/**
+ * Helper function to centralize the calling of methods based on the passed replica selection
+ * mechanism.
+ * @param replicaSelection replica selection mechanism to use
+ * @return a UUID for the server that matches the selection, can be null
+ */
+ String getReplicaSelectedUUID(ReplicaSelection replicaSelection) {
+ switch (replicaSelection) {
+ case LEADER_ONLY:
+ return getLeaderUUID();
+ case CLOSEST_REPLICA:
+ return getClosestUUID();
+ default:
+ throw new RuntimeException("Unknown replica selection mechanism " + replicaSelection);
+ }
+ }
+
+ /**
* Gets the replicas of this tablet. The returned list may not be mutated.
* @return the replicas of the tablet
*/
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java
new file mode 100644
index 0000000..e33ba34
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java
@@ -0,0 +1,36 @@
+// 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.kudu.annotations.InterfaceAudience;
+import org.apache.kudu.annotations.InterfaceStability;
+
+/**
+ * Policy with which to choose amongst multiple replicas.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public enum ReplicaSelection {
+ /**
+ * Select the LEADER replica.
+ */
+ LEADER_ONLY,
+ /**
+ * Select the closest replica to the client, or a random one if all replicas are equidistant.
+ */
+ CLOSEST_REPLICA
+}
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
index aa50ad6..ea77c27 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
@@ -101,6 +101,14 @@ public class TestRemoteTablet {
assertNotNull(tablet.getClosestUUID());
}
+ @Test
+ public void testReplicaSelection() {
+ RemoteTablet tablet = getTablet(0, 1);
+
+ assertEquals("0", tablet.getReplicaSelectedUUID(ReplicaSelection.LEADER_ONLY));
+ assertEquals("1", tablet.getReplicaSelectedUUID(ReplicaSelection.CLOSEST_REPLICA));
+ }
+
private RemoteTablet getTablet(int leaderIndex) {
return getTablet(leaderIndex, -1);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/a7fd4fc8/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
index 71739d4..14454c0 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
@@ -176,6 +176,21 @@ public class TestScannerMultiTablet extends BaseKuduTest {
buildScannerAndCheckColumnsCount(builder, 2);
}
+ @Test(timeout = 100000)
+ public void testReplicaSelections() throws Exception {
+ AsyncKuduScanner scanner = client.newScannerBuilder(table)
+ .replicaSelection(ReplicaSelection.LEADER_ONLY)
+ .build();
+
+ assertEquals(9, countRowsInScan(scanner));
+
+ scanner = client.newScannerBuilder(table)
+ .replicaSelection(ReplicaSelection.CLOSEST_REPLICA)
+ .build();
+
+ assertEquals(9, countRowsInScan(scanner));
+ }
+
private AsyncKuduScanner getScanner(String lowerBoundKeyOne,
String lowerBoundKeyTwo,
String exclusiveUpperBoundKeyOne,
[5/5] kudu git commit: [java client] KuduRpc.deserialize should throw
KuduException
Posted by jd...@apache.org.
[java client] KuduRpc.deserialize should throw KuduException
A vestige of our use of Exceptions everywhere remained in KuduRpc where we
were sending them in deserialize. This change helps with the tracing work.
Change-Id: I8196e949d4686113c0cdd7b611b14898df9b3bb7
Reviewed-on: http://gerrit.cloudera.org:8080/4872
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/b333a36f
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/b333a36f
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/b333a36f
Branch: refs/heads/master
Commit: b333a36f12d019dbecb734cf2020d5f6f71fa26b
Parents: a7fd4fc
Author: Jean-Daniel Cryans <jd...@apache.org>
Authored: Thu Oct 27 10:44:51 2016 -0700
Committer: Jean-Daniel Cryans <jd...@apache.org>
Committed: Thu Oct 27 23:07:45 2016 +0000
----------------------------------------------------------------------
.../apache/kudu/client/AlterTableRequest.java | 2 +-
.../org/apache/kudu/client/AsyncKuduClient.java | 2 +-
.../apache/kudu/client/AsyncKuduScanner.java | 2 +-
.../main/java/org/apache/kudu/client/Batch.java | 3 +--
.../apache/kudu/client/CreateTableRequest.java | 2 +-
.../apache/kudu/client/DeleteTableRequest.java | 2 +-
.../client/GetMasterRegistrationRequest.java | 2 +-
.../kudu/client/GetTableLocationsRequest.java | 2 +-
.../kudu/client/GetTableSchemaRequest.java | 2 +-
.../kudu/client/IsAlterTableDoneRequest.java | 2 +-
.../kudu/client/IsCreateTableDoneRequest.java | 2 +-
.../java/org/apache/kudu/client/KuduRpc.java | 13 ++++++-----
.../apache/kudu/client/ListTablesRequest.java | 2 +-
.../kudu/client/ListTabletServersRequest.java | 2 +-
.../apache/kudu/client/ListTabletsRequest.java | 4 +---
.../java/org/apache/kudu/client/Operation.java | 3 +--
.../org/apache/kudu/client/PingRequest.java | 2 +-
.../org/apache/kudu/client/TabletClient.java | 24 ++++++++++----------
18 files changed, 35 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
index 41eb1a1..8e6cb7f 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
@@ -66,7 +66,7 @@ class AlterTableRequest extends KuduRpc<AlterTableResponse> {
@Override
Pair<AlterTableResponse, Object> deserialize(final CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final AlterTableResponsePB.Builder respBuilder = AlterTableResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
AlterTableResponse response = new AlterTableResponse(
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/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 35cd9d6..74de993 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
@@ -420,7 +420,7 @@ public class AsyncKuduClient implements AutoCloseable {
@Override
Pair<KuduTable, Object> deserialize(CallResponse callResponse, String tsUUID)
- throws Exception { return null; }
+ throws KuduException { return null; }
};
fakeRpc.setTimeoutMillis(defaultAdminOperationTimeoutMs);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/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 b51394a..3864088 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
@@ -751,7 +751,7 @@ public final class AsyncKuduScanner {
@Override
Pair<Response, Object> deserialize(final CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
ScanResponsePB.Builder builder = ScanResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
ScanResponsePB resp = builder.build();
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java b/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
index 01bc113..8dacdd7 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
@@ -33,7 +33,6 @@ import org.apache.kudu.client.Statistics.TabletStatistics;
import org.apache.kudu.tserver.Tserver;
import org.apache.kudu.tserver.Tserver.TabletServerErrorPB;
import org.apache.kudu.util.Pair;
-import org.apache.kudu.util.Slice;
/**
* Used internally to group Operations for a single tablet together before sending to the tablet
@@ -108,7 +107,7 @@ class Batch extends KuduRpc<BatchResponse> {
@Override
Pair<BatchResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
Tserver.WriteResponsePB.Builder builder = Tserver.WriteResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/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 6dc00cb..59aefa7 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
@@ -67,7 +67,7 @@ class CreateTableRequest extends KuduRpc<CreateTableResponse> {
@Override
Pair<CreateTableResponse, Object> deserialize(final CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final Master.CreateTableResponsePB.Builder builder = Master.CreateTableResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
CreateTableResponse response =
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
index 8060d54..dda31f6 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
@@ -57,7 +57,7 @@ class DeleteTableRequest extends KuduRpc<DeleteTableResponse> {
@Override
Pair<DeleteTableResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final Master.DeleteTableResponsePB.Builder builder = Master.DeleteTableResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
DeleteTableResponse response =
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/GetMasterRegistrationRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/GetMasterRegistrationRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/GetMasterRegistrationRequest.java
index 2e76e30..c28e792 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/GetMasterRegistrationRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/GetMasterRegistrationRequest.java
@@ -53,7 +53,7 @@ public class GetMasterRegistrationRequest extends KuduRpc<GetMasterRegistrationR
@Override
Pair<GetMasterRegistrationResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final GetMasterRegistrationResponsePB.Builder respBuilder =
GetMasterRegistrationResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
index 1bc2a14..b535d6d 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
@@ -58,7 +58,7 @@ class GetTableLocationsRequest extends KuduRpc<Master.GetTableLocationsResponseP
@Override
Pair<Master.GetTableLocationsResponsePB, Object> deserialize(
final CallResponse callResponse, String tsUUID)
- throws Exception {
+ throws KuduException {
Master.GetTableLocationsResponsePB.Builder builder = Master.GetTableLocationsResponsePB
.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/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 d76b24a..11a8405 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
@@ -58,7 +58,7 @@ public class GetTableSchemaRequest extends KuduRpc<GetTableSchemaResponse> {
@Override
Pair<GetTableSchemaResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final GetTableSchemaResponsePB.Builder respBuilder = GetTableSchemaResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
Schema schema = ProtobufHelper.pbToSchema(respBuilder.getSchema());
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/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 d923984..dd42fe0 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
@@ -58,7 +58,7 @@ class IsAlterTableDoneRequest extends KuduRpc<IsAlterTableDoneResponse> {
@Override
Pair<IsAlterTableDoneResponse, Object> deserialize(final CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final IsAlterTableDoneResponsePB.Builder respBuilder = IsAlterTableDoneResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
IsAlterTableDoneResponse resp = new IsAlterTableDoneResponse(deadlineTracker.getElapsedMillis(),
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/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 61b9ef6..bfd6db9 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
@@ -46,7 +46,7 @@ class IsCreateTableDoneRequest extends KuduRpc<Master.IsCreateTableDoneResponseP
@Override
Pair<Master.IsCreateTableDoneResponsePB, Object> deserialize(
- final CallResponse callResponse, String tsUUID) throws Exception {
+ final CallResponse callResponse, String tsUUID) throws KuduException {
Master.IsCreateTableDoneResponsePB.Builder builder = Master.IsCreateTableDoneResponsePB
.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
index 8b5c0df..5ae3f56 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
@@ -149,14 +149,15 @@ public abstract class KuduRpc<R> {
* Notice that this method is package-private, so only classes within this
* package can use this as a base class.
*
- * @param callResponse The call response from which to deserialize.
- * @param tsUUID A string that contains the UUID of the server that answered the RPC.
- * @return An Object of type R that will be sent to callback and an Object that will be an Error
+ * @param callResponse the call response from which to deserialize
+ * @param tsUUID a string that contains the UUID of the server that answered the RPC
+ * @return an Object of type R that will be sent to callback and an Object that will be an Error
* of type TabletServerErrorPB or MasterErrorPB that will be converted into an exception and
- * sent to errback.
- * @throws Exception An exception that will be sent to errback.
+ * sent to errback
+ * @throws KuduException an exception that will be sent to errback
*/
- abstract Pair<R, Object> deserialize(CallResponse callResponse, String tsUUID) throws Exception;
+ abstract Pair<R, Object> deserialize(CallResponse callResponse, String tsUUID)
+ throws KuduException;
/**
* Update the statistics information before this rpc is called back. This method should not throw
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
index 61b27c8..470f45d 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
@@ -56,7 +56,7 @@ class ListTablesRequest extends KuduRpc<ListTablesResponse> {
@Override
Pair<ListTablesResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final Master.ListTablesResponsePB.Builder respBuilder =
Master.ListTablesResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
index 4c6a7c9..c16c223 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
@@ -50,7 +50,7 @@ public class ListTabletServersRequest extends KuduRpc<ListTabletServersResponse>
@Override
Pair<ListTabletServersResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final ListTabletServersResponsePB.Builder respBuilder =
ListTabletServersResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
index f8b835d..1fe3d4d 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
@@ -18,9 +18,7 @@ package org.apache.kudu.client;
import com.google.protobuf.Message;
import org.apache.kudu.annotations.InterfaceAudience;
-import org.apache.kudu.tablet.Tablet;
import org.apache.kudu.tserver.Tserver;
-import org.apache.kudu.tserver.TserverService;
import org.apache.kudu.util.Pair;
import org.jboss.netty.buffer.ChannelBuffer;
@@ -52,7 +50,7 @@ class ListTabletsRequest extends KuduRpc<ListTabletsResponse> {
@Override
Pair<ListTabletsResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
final Tserver.ListTabletsResponsePB.Builder respBuilder =
Tserver.ListTabletsResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
index f0bd71a..512bddd 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
@@ -32,7 +32,6 @@ import org.apache.kudu.client.Statistics.Statistic;
import org.apache.kudu.client.Statistics.TabletStatistics;
import org.apache.kudu.tserver.Tserver;
import org.apache.kudu.util.Pair;
-import org.apache.kudu.util.Slice;
import org.jboss.netty.buffer.ChannelBuffer;
import java.nio.ByteBuffer;
@@ -139,7 +138,7 @@ public abstract class Operation extends KuduRpc<OperationResponse> {
@Override
Pair<OperationResponse, Object> deserialize(CallResponse callResponse,
- String tsUUID) throws Exception {
+ String tsUUID) throws KuduException {
Tserver.WriteResponsePB.Builder builder = Tserver.WriteResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
Tserver.WriteResponsePB.PerRowErrorPB error = null;
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
index 31a3f2f..79e788b 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
@@ -64,7 +64,7 @@ class PingRequest extends KuduRpc<PingResponse> {
}
@Override
- Pair<PingResponse, Object> deserialize(CallResponse callResponse, String tsUUID) throws Exception {
+ Pair<PingResponse, Object> deserialize(CallResponse callResponse, String tsUUID) throws KuduException {
final Master.PingResponsePB.Builder respBuilder =
Master.PingResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
http://git-wip-us.apache.org/repos/asf/kudu/blob/b333a36f/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
index 822fac6..5ff5e98 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
@@ -421,7 +421,7 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
}
Pair<Object, Object> decoded = null;
- Exception exception = null;
+ KuduException exception = null;
Status retryableHeaderError = Status.OK();
if (header.hasIsError() && header.getIsError()) {
RpcHeader.ErrorStatusPB.Builder errorBuilder = RpcHeader.ErrorStatusPB.newBuilder();
@@ -441,7 +441,7 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
} else {
try {
decoded = rpc.deserialize(response, this.serverInfo.getUuid());
- } catch (Exception ex) {
+ } catch (KuduException ex) {
exception = ex;
}
}
@@ -521,12 +521,12 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
/**
* Takes care of a few kinds of TS errors that we handle differently, like tablets or leaders
* moving. Builds and returns an exception if we don't know what to do with it.
- * @param rpc The original RPC call that triggered the error.
- * @param error The error the TS sent.
- * @return An exception if we couldn't dispatch the error, or null.
+ * @param rpc the original RPC call that triggered the error
+ * @param error the error the TS sent
+ * @return an exception if we couldn't dispatch the error, or null
*/
- private Exception dispatchTSErrorOrReturnException(KuduRpc rpc,
- Tserver.TabletServerErrorPB error) {
+ private KuduException dispatchTSErrorOrReturnException(KuduRpc rpc,
+ Tserver.TabletServerErrorPB error) {
WireProtocol.AppStatusPB.ErrorCode code = error.getStatus().getCode();
Status status = Status.fromTabletServerErrorPB(error);
if (error.getCode() == Tserver.TabletServerErrorPB.Code.TABLET_NOT_FOUND) {
@@ -547,12 +547,12 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
/**
* Provides different handling for various kinds of master errors: re-uses the
* mechanisms already in place for handling tablet server errors as much as possible.
- * @param rpc The original RPC call that triggered the error.
- * @param error The error the master sent.
- * @return An exception if we couldn't dispatch the error, or null.
+ * @param rpc the original RPC call that triggered the error
+ * @param error the error the master sent
+ * @return an exception if we couldn't dispatch the error, or null
*/
- private Exception dispatchMasterErrorOrReturnException(KuduRpc rpc,
- Master.MasterErrorPB error) {
+ private KuduException dispatchMasterErrorOrReturnException(KuduRpc rpc,
+ Master.MasterErrorPB error) {
WireProtocol.AppStatusPB.ErrorCode code = error.getStatus().getCode();
Status status = Status.fromMasterErrorPB(error);
if (error.getCode() == Master.MasterErrorPB.Code.NOT_THE_LEADER) {
[3/5] kudu git commit: [java client] Refactor all server info into a
single class, add locality
Posted by jd...@apache.org.
[java client] Refactor all server info into a single class, add locality
Having this helps propagate information around the client. RemoteTablet can now
answer requests for local servers.
This patch also tries to simplify how RemoteTablets are created, as it's getting
a bit messy and redundant with the passing of ServerInfos.
Change-Id: I33984a437d8c8d07d5db4d16f8da723b3e904189
Reviewed-on: http://gerrit.cloudera.org:8080/4836
Tested-by: Kudu Jenkins
Reviewed-by: David Ribeiro Alves <dr...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/a326fc95
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/a326fc95
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/a326fc95
Branch: refs/heads/master
Commit: a326fc950681c7187c096b5899186932e6e8d080
Parents: 96c2e2a
Author: Jean-Daniel Cryans <jd...@apache.org>
Authored: Mon Oct 24 10:17:36 2016 -0700
Committer: Jean-Daniel Cryans <jd...@apache.org>
Committed: Thu Oct 27 18:18:13 2016 +0000
----------------------------------------------------------------------
.../org/apache/kudu/client/AsyncKuduClient.java | 19 ++---
.../org/apache/kudu/client/ConnectionCache.java | 27 ++++---
.../org/apache/kudu/client/RemoteTablet.java | 54 +++++++++-----
.../java/org/apache/kudu/client/ServerInfo.java | 78 ++++++++++++++++++++
.../org/apache/kudu/client/TabletClient.java | 56 +++-----------
.../apache/kudu/client/TestConnectionCache.java | 3 +-
.../apache/kudu/client/TestRemoteTablet.java | 29 +++++++-
7 files changed, 178 insertions(+), 88 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/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 a781f9f..b36737b 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
@@ -1137,7 +1137,7 @@ public class AsyncKuduClient implements AutoCloseable {
* a RPC, so we need to demote it and retry.
*/
<R> void handleNotLeader(final KuduRpc<R> rpc, KuduException ex, TabletClient server) {
- rpc.getTablet().demoteLeader(server.getUuid());
+ rpc.getTablet().demoteLeader(server.getServerInfo().getUuid());
handleRetryableError(rpc, ex);
}
@@ -1182,9 +1182,9 @@ public class AsyncKuduClient implements AutoCloseable {
* the tablet itself from the caches.
*/
private void invalidateTabletCache(RemoteTablet tablet, TabletClient server) {
- LOG.info("Removing server " + server.getUuid() + " from this tablet's cache " +
- tablet.getTabletId());
- tablet.removeTabletClient(server.getUuid());
+ String uuid = server.getServerInfo().getUuid();
+ LOG.info("Removing server {} from this tablet's cache {}", uuid, tablet.getTabletId());
+ tablet.removeTabletClient(uuid);
}
/** Callback executed when a master lookup completes. */
@@ -1275,12 +1275,14 @@ public class AsyncKuduClient implements AutoCloseable {
List<RemoteTablet> tablets = new ArrayList<>(locations.size());
for (Master.TabletLocationsPB tabletPb : locations) {
- String tabletId = tabletPb.getTabletId().toStringUtf8();
-
List<UnknownHostException> lookupExceptions = new ArrayList<>(tabletPb.getReplicasCount());
+ List<ServerInfo> servers = new ArrayList<>(tabletPb.getReplicasCount());
for (Master.TabletLocationsPB.ReplicaPB replica : tabletPb.getReplicasList()) {
try {
- connectionCache.connectTS(replica.getTsInfo());
+ ServerInfo serverInfo = connectionCache.connectTS(replica.getTsInfo());
+ if (serverInfo != null) {
+ servers.add(serverInfo);
+ }
} catch (UnknownHostException ex) {
lookupExceptions.add(ex);
}
@@ -1293,8 +1295,7 @@ public class AsyncKuduClient implements AutoCloseable {
throw new NonRecoverableException(statusIOE);
}
- Partition partition = ProtobufHelper.pbToPartition(tabletPb.getPartition());
- RemoteTablet rt = new RemoteTablet(tableId, tabletId, partition, tabletPb);
+ RemoteTablet rt = new RemoteTablet(tableId, tabletPb, servers);
LOG.info("Learned about tablet {} for table '{}' with partition {}",
rt.getTabletId(), tableName, rt.getPartition());
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectionCache.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectionCache.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectionCache.java
index ed6100e..9d8f1d0 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectionCache.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectionCache.java
@@ -92,14 +92,15 @@ class ConnectionCache {
/**
* Create a connection to a tablet server based on information provided by the master.
* @param tsInfoPB master-provided information for the tablet server
+ * @return an object that contains all the server's information
* @throws UnknownHostException if we cannot resolve the tablet server's IP address
*/
- void connectTS(Master.TSInfoPB tsInfoPB) throws UnknownHostException {
+ ServerInfo connectTS(Master.TSInfoPB tsInfoPB) throws UnknownHostException {
List<Common.HostPortPB> addresses = tsInfoPB.getRpcAddressesList();
String uuid = tsInfoPB.getPermanentUuid().toStringUtf8();
if (addresses.isEmpty()) {
LOG.warn("Received a tablet server with no addresses, UUID: {}", uuid);
- return;
+ return null;
}
// from meta_cache.cc
@@ -110,29 +111,30 @@ class ConnectionCache {
throw new UnknownHostException(
"Failed to resolve the IP of `" + addresses.get(0).getHost() + "'");
}
- newClient(uuid, inetAddress, addresses.get(0).getPort());
+ return newClient(uuid, inetAddress, addresses.get(0).getPort()).getServerInfo();
}
TabletClient newClient(String uuid, InetAddress inetAddress, int port) {
String host = inetAddress.getHostAddress();
boolean isLocal = NetUtil.isLocalAddress(inetAddress);
- return newClient(uuid, host, port, isLocal);
+ ServerInfo serverInfo = new ServerInfo(uuid, host, port, isLocal);
+ return newClient(serverInfo);
}
- TabletClient newClient(String uuid, String host, int port, boolean isLocal) {
+ TabletClient newClient(ServerInfo serverInfo) {
TabletClient client;
SocketChannel chan;
writeLock.lock();
try {
- client = uuid2client.get(uuid);
+ client = uuid2client.get(serverInfo.getUuid());
if (client != null && client.isAlive()) {
return client;
}
final TabletClientPipeline pipeline = new TabletClientPipeline();
- client = pipeline.init(uuid, host, port, isLocal);
+ client = pipeline.init(serverInfo);
chan = this.kuduClient.getChannelFactory().newChannel(pipeline);
- uuid2client.put(uuid, client);
+ uuid2client.put(serverInfo.getUuid(), client);
} finally {
writeLock.unlock();
}
@@ -144,7 +146,8 @@ class ConnectionCache {
// Java since the JRE doesn't expose any way to call setsockopt() with
// TCP_KEEPIDLE. And of course the default timeout is >2h. Sigh.
config.setKeepAlive(true);
- chan.connect(new InetSocketAddress(host, port)); // Won't block.
+ chan.connect(
+ new InetSocketAddress(serverInfo.getHostname(), serverInfo.getPort())); // Won't block.
return client;
}
@@ -178,7 +181,7 @@ class ConnectionCache {
} else if (client.isAlive()) {
return client;
} else {
- return newClient(uuid, client.getHost(), client.getPort(), client.isLocal());
+ return newClient(client.getServerInfo());
}
}
@@ -235,9 +238,9 @@ class ConnectionCache {
private final class TabletClientPipeline extends DefaultChannelPipeline {
- TabletClient init(String uuid, String host, int port, boolean isLocal) {
+ TabletClient init(ServerInfo serverInfo) {
AsyncKuduClient kuduClient = ConnectionCache.this.kuduClient;
- final TabletClient client = new TabletClient(kuduClient, uuid, host, port, isLocal);
+ final TabletClient client = new TabletClient(kuduClient, serverInfo);
if (kuduClient.getDefaultSocketReadTimeoutMs() > 0) {
super.addLast("timeout-handler",
new ReadTimeoutHandler(kuduClient.getTimer(),
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
index 07936c8..28be0fe 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
@@ -27,16 +27,15 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.concurrent.GuardedBy;
-import java.util.ArrayList;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
-import java.util.Set;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
/**
* This class encapsulates the information regarding a tablet and its locations.
* <p>
- * RemoteTablet's main function, once it is init()'d, is to keep track of where the leader for this
+ * RemoteTablet's main function is to keep track of where the leader for this
* tablet is. For example, an RPC might call {@link #getLeaderUUID()}, contact that TS, find
* it's not the leader anymore, and then call {@link #demoteLeader(String)}.
* <p>
@@ -53,7 +52,7 @@ class RemoteTablet implements Comparable<RemoteTablet> {
private final String tableId;
private final String tabletId;
@GuardedBy("tabletServers")
- private final Set<String> tabletServers = new HashSet<>();
+ private final Map<String, ServerInfo> tabletServers;
private final AtomicReference<List<LocatedTablet.Replica>> replicas =
new AtomicReference(ImmutableList.of());
private final Partition partition;
@@ -61,15 +60,22 @@ class RemoteTablet implements Comparable<RemoteTablet> {
@GuardedBy("tabletServers")
private String leaderUuid;
- RemoteTablet(String tableId, String tabletId,
- Partition partition, Master.TabletLocationsPB tabletLocations) {
- this.tabletId = tabletId;
+ RemoteTablet(String tableId,
+ Master.TabletLocationsPB tabletLocations,
+ List<ServerInfo> serverInfos) {
+ this.tabletId = tabletLocations.getTabletId().toStringUtf8();
this.tableId = tableId;
- this.partition = partition;
+ this.partition = ProtobufHelper.pbToPartition(tabletLocations.getPartition());
+ this.tabletServers = new HashMap<>(serverInfos.size());
+ for (ServerInfo serverInfo : serverInfos) {
+ this.tabletServers.put(serverInfo.getUuid(), serverInfo);
+ }
+
+ ImmutableList.Builder<LocatedTablet.Replica> replicasBuilder = new ImmutableList.Builder<>();
for (Master.TabletLocationsPB.ReplicaPB replica : tabletLocations.getReplicasList()) {
String uuid = replica.getTsInfo().getPermanentUuid().toStringUtf8();
- tabletServers.add(uuid);
+ replicasBuilder.add(new LocatedTablet.Replica(replica));
if (replica.getRole().equals(Metadata.RaftPeerPB.Role.LEADER)) {
leaderUuid = uuid;
}
@@ -78,11 +84,6 @@ class RemoteTablet implements Comparable<RemoteTablet> {
if (leaderUuid == null) {
LOG.warn("No leader provided for tablet {}", getTabletId());
}
-
- ImmutableList.Builder<LocatedTablet.Replica> replicasBuilder = new ImmutableList.Builder<>();
- for (Master.TabletLocationsPB.ReplicaPB replica : tabletLocations.getReplicasList()) {
- replicasBuilder.add(new LocatedTablet.Replica(replica));
- }
replicas.set(replicasBuilder.build());
}
@@ -101,7 +102,7 @@ class RemoteTablet implements Comparable<RemoteTablet> {
if (leaderUuid != null && leaderUuid.equals(uuid)) {
leaderUuid = null;
}
- if (tabletServers.remove(uuid)) {
+ if (tabletServers.remove(uuid) != null) {
return true;
}
LOG.debug("tablet {} already removed ts {}, size left is {}",
@@ -135,7 +136,7 @@ class RemoteTablet implements Comparable<RemoteTablet> {
}
/**
- * Get the UUID of the tablet server that we think holds the leader replica for this tablet.
+ * Gets the UUID of the tablet server that we think holds the leader replica for this tablet.
* @return a UUID of a tablet server that we think has the leader, else null
*/
String getLeaderUUID() {
@@ -145,6 +146,25 @@ class RemoteTablet implements Comparable<RemoteTablet> {
}
/**
+ * Gets the UUID of the closest server. If none is closer than the others, returns a random
+ * server UUID.
+ * @return the UUID of the closest server, which might be any if none is closer, or null if this
+ * cache doesn't know of any servers
+ */
+ String getClosestUUID() {
+ synchronized (tabletServers) {
+ String lastUuid = null;
+ for (ServerInfo serverInfo : tabletServers.values()) {
+ lastUuid = serverInfo.getUuid();
+ if (serverInfo.isLocal()) {
+ return serverInfo.getUuid();
+ }
+ }
+ return lastUuid;
+ }
+ }
+
+ /**
* Gets the replicas of this tablet. The returned list may not be mutated.
* @return the replicas of the tablet
*/
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
new file mode 100644
index 0000000..999f7f2
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
@@ -0,0 +1,78 @@
+// 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.kudu.annotations.InterfaceAudience;
+
+/**
+ * Container class for server information that never changes, like UUID and hostname.
+ */
+@InterfaceAudience.Private
+public class ServerInfo {
+ private final String uuid;
+ private final String hostname;
+ private final int port;
+ private final boolean local;
+
+ /**
+ * Constructor for all the fields. The intent is that there should only be one ServerInfo
+ * instance per UUID the client is connected to.
+ * @param uuid server's UUID
+ * @param hostname server's hostname, only one of them
+ * @param port server's port
+ * @param local if the server is hosted on the same machine where this client is running
+ */
+ public ServerInfo(String uuid, String hostname, int port, boolean local) {
+ this.uuid = uuid;
+ this.hostname = hostname;
+ this.port = port;
+ this.local = local;
+ }
+
+ /**
+ * Returns this server's uuid.
+ * @return a string that contains this server's uuid
+ */
+ public String getUuid() {
+ return uuid;
+ }
+
+ /**
+ * Returns this server's hostname. We might get many hostnames from the master for a single
+ * TS, and this is the one we picked to connect to originally.
+ * @return a string that contains this server's hostname
+ */
+ public String getHostname() {
+ return hostname;
+ }
+
+ /**
+ * Returns this server's port.
+ * @return a port number that this server is bound to
+ */
+ public int getPort() {
+ return port;
+ }
+
+ /**
+ * Returns if this server is on this client's host.
+ * @return true if the server is local, else false
+ */
+ public boolean isLocal() {
+ return local;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
index 0b190ed..822fac6 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/TabletClient.java
@@ -133,32 +133,23 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
private final AsyncKuduClient kuduClient;
- private final String uuid;
-
- private final String host;
-
- private final int port;
-
private final long socketReadTimeoutMs;
private SecureRpcHelper secureRpcHelper;
private final RequestTracker requestTracker;
+ private final ServerInfo serverInfo;
+
// If an uncaught exception forced us to shutdown this TabletClient, we'll handle the retry
// differently by also clearing the caches.
private volatile boolean gotUncaughtException = false;
- private final boolean local;
-
- public TabletClient(AsyncKuduClient client, String uuid, String host, int port, boolean local) {
+ public TabletClient(AsyncKuduClient client, ServerInfo serverInfo) {
this.kuduClient = client;
- this.uuid = uuid;
this.socketReadTimeoutMs = client.getDefaultSocketReadTimeoutMs();
- this.host = host;
- this.port = port;
this.requestTracker = client.getRequestTracker();
- this.local = local;
+ this.serverInfo = serverInfo;
}
<R> void sendRpc(KuduRpc<R> rpc) {
@@ -449,7 +440,7 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
}
} else {
try {
- decoded = rpc.deserialize(response, this.uuid);
+ decoded = rpc.deserialize(response, this.serverInfo.getUuid());
} catch (Exception ex) {
exception = ex;
}
@@ -832,40 +823,11 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
}
private String getPeerUuidLoggingString() {
- return "[Peer " + uuid + "] ";
+ return "[Peer " + serverInfo.getUuid() + "] ";
}
- /**
- * Returns this tablet server's uuid.
- * @return a string that contains this tablet server's uuid
- */
- String getUuid() {
- return uuid;
- }
-
- /**
- * Returns if this server is on this client's host.
- * @return true if the server is local, else false
- */
- boolean isLocal() {
- return local;
- }
-
- /**
- * Returns this tablet server's port.
- * @return a port number that this tablet server is bound to
- */
- int getPort() {
- return port;
- }
-
- /**
- * Returns this tablet server's hostname. We might get many hostnames from the master for a single
- * TS, and this is the one we picked to connect to originally.
- * @returna string that contains this tablet server's hostname
- */
- String getHost() {
- return host;
+ ServerInfo getServerInfo() {
+ return serverInfo;
}
public String toString() {
@@ -875,7 +837,7 @@ public class TabletClient extends ReplayingDecoder<VoidEnum> {
.append("(chan=") // = 6
.append(chan) // ~64 (up to 66 when using IPv4)
.append(", uuid=") // = 7
- .append(uuid) // = 32
+ .append(serverInfo.getUuid()) // = 32
.append(", #pending_rpcs="); // =16
int npending_rpcs;
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
index e238069..334dd38 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
@@ -39,7 +39,8 @@ public class TestConnectionCache {
ConnectionCache cache = new ConnectionCache(client);
int i = 0;
for (HostAndPort hp : addresses) {
- TabletClient conn = cache.newClient(i + "", hp.getHostText(), hp.getPort(), false);
+ TabletClient conn =
+ cache.newClient(new ServerInfo(i + "", hp.getHostText(), hp.getPort(), false));
// Ping the process so we go through the whole connection process.
pingConnection(conn);
i++;
http://git-wip-us.apache.org/repos/asf/kudu/blob/a326fc95/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
index 5cf4de8..aa50ad6 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
@@ -21,6 +21,9 @@ import org.apache.kudu.consensus.Metadata;
import org.apache.kudu.master.Master;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.List;
+
import static org.junit.Assert.*;
public class TestRemoteTablet {
@@ -83,17 +86,39 @@ public class TestRemoteTablet {
assertTrue(tablet.removeTabletClient("2"));
}
+ @Test
+ public void testLocalReplica() {
+ RemoteTablet tablet = getTablet(0, 0);
+
+ assertEquals("0", tablet.getClosestUUID());
+ }
+
+ @Test
+ public void testNoLocalReplica() {
+ RemoteTablet tablet = getTablet(0, -1);
+
+ // We just care about getting one back.
+ assertNotNull(tablet.getClosestUUID());
+ }
+
private RemoteTablet getTablet(int leaderIndex) {
+ return getTablet(leaderIndex, -1);
+ }
+
+ private RemoteTablet getTablet(int leaderIndex, int localReplicaIndex) {
Master.TabletLocationsPB.Builder tabletPb = Master.TabletLocationsPB.newBuilder();
tabletPb.setPartition(TestUtils.getFakePartitionPB());
tabletPb.setTabletId(ByteString.copyFromUtf8("fake tablet"));
+ List<ServerInfo> servers = new ArrayList<>();
for (int i = 0; i < 3; i++) {
+ String uuid = i + "";
+ servers.add(new ServerInfo(uuid, "host", i, i == localReplicaIndex));
tabletPb.addReplicas(TestUtils.getFakeTabletReplicaPB(
- i + "", "host", i,
+ uuid, "host", i,
leaderIndex == i ? Metadata.RaftPeerPB.Role.LEADER : Metadata.RaftPeerPB.Role.FOLLOWER));
}
- return new RemoteTablet("fake table", "fake tablet", null, tabletPb.build());
+ return new RemoteTablet("fake table", tabletPb.build(), servers);
}
}
[2/5] kudu git commit: Redirect krb5 stdout to SLF4j
Posted by jd...@apache.org.
Redirect krb5 stdout to SLF4j
Change-Id: Iaf49340cbe5f0630f2e0674274b94c853c4ccfc3
Reviewed-on: http://gerrit.cloudera.org:8080/4843
Reviewed-by: Todd Lipcon <to...@apache.org>
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/96c2e2a8
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/96c2e2a8
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/96c2e2a8
Branch: refs/heads/master
Commit: 96c2e2a83cf4a371e4ca179248bb05773359e985
Parents: f2a39c6
Author: Dan Burkert <da...@apache.org>
Authored: Tue Oct 25 14:53:45 2016 -0700
Committer: Jean-Daniel Cryans <jd...@apache.org>
Committed: Thu Oct 27 15:40:55 2016 +0000
----------------------------------------------------------------------
.../java/org/apache/kudu/client/MiniKdc.java | 78 ++++++++++++++------
.../org/apache/kudu/client/MiniKuduCluster.java | 3 +-
2 files changed, 56 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/96c2e2a8/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKdc.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKdc.java b/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKdc.java
index ecd7041..53ae002 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKdc.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKdc.java
@@ -25,9 +25,11 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.io.CharStreams;
import org.apache.commons.io.FileUtils;
+import org.apache.kudu.annotations.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.BufferedReader;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
@@ -39,6 +41,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import javax.annotation.concurrent.NotThreadSafe;
/**
* A managed Kerberos Key Distribution Center.
@@ -48,6 +51,8 @@ import java.util.Map;
*
* The KDC is managed as an external process, using the krb5 binaries installed on the system.
*/
+@InterfaceAudience.Private
+@NotThreadSafe
public class MiniKdc implements Closeable {
// The KDC port will be assigned starting from this value.
@@ -57,7 +62,9 @@ public class MiniKdc implements Closeable {
private final Options options;
- Process kdcProcess;
+ private Process kdcProcess;
+
+ private Thread kdcProcessLogRedirector;
/**
* Options for the MiniKdc.
@@ -145,12 +152,19 @@ public class MiniKdc implements Closeable {
"-s", // Stash the master password.
"-P", "masterpw", // Set a password.
"-W" // Use weak entropy (since we don't need real security).
- ), "kdb5_util");
+ ), "kdb5_util", true);
}
kdcProcess = startProcessWithKrbEnv(getBinaryPath("krb5kdc"),
"-n"); // Do not daemonize.
+ // Redirect the KDC output to SLF4J.
+ kdcProcessLogRedirector = new Thread(
+ new MiniKuduCluster.ProcessInputStreamLogPrinterRunnable(kdcProcess.getInputStream()),
+ "krb5kdc:" + options.port);
+ kdcProcessLogRedirector.setDaemon(true);
+ kdcProcessLogRedirector.start();
+
// The C++ MiniKdc defaults to binding the KDC to an ephemeral port, which
// it then finds using lsof at this point. Java is unable to do that since
// the Process API does not expose the subprocess PID. As a result, this
@@ -169,7 +183,7 @@ public class MiniKdc implements Closeable {
getBinaryPath("kadmin.local"),
"-q",
String.format("add_principal -pw %s %s", username, username)
- ), "kadmin.local");
+ ), "kadmin.local", true);
}
/**
@@ -180,7 +194,7 @@ public class MiniKdc implements Closeable {
Process proc = startProcessWithKrbEnv(getBinaryPath("kinit"), username);
proc.getOutputStream().write(username.getBytes());
proc.getOutputStream().close();
- checkReturnCode(proc, "kinit");
+ checkReturnCode(proc, "kinit", true);
}
/**
@@ -188,9 +202,8 @@ public class MiniKdc implements Closeable {
* local ticket cache state.
*/
String klist() throws IOException {
- Process proc = buildProcessWithKrbEnv(getBinaryPath("klist"), "-A")
- .redirectOutput(ProcessBuilder.Redirect.PIPE).start();
- checkReturnCode(proc, "klist");
+ Process proc = startProcessWithKrbEnv(getBinaryPath("klist"), "-A");
+ checkReturnCode(proc, "klist", false);
return CharStreams.toString(new InputStreamReader(proc.getInputStream()));
}
@@ -207,12 +220,12 @@ public class MiniKdc implements Closeable {
checkReturnCode(startProcessWithKrbEnv(kadmin,
"-q",
String.format("add_principal -randkey %s", spn)),
- "kadmin.local");
+ "kadmin.local", true);
checkReturnCode(startProcessWithKrbEnv(kadmin,
"-q",
String.format("ktadd -k %s %s", kt_path, spn)),
- "kadmin.local");
+ "kadmin.local", true);
return kt_path;
}
@@ -268,10 +281,12 @@ public class MiniKdc implements Closeable {
try {
kdcProcess.destroy();
kdcProcess.waitFor();
+ kdcProcessLogRedirector.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
kdcProcess = null;
+ kdcProcessLogRedirector = null;
}
}
@@ -305,11 +320,7 @@ public class MiniKdc implements Closeable {
);
}
- private static String getBinaryPath(String executable) throws IOException {
- return getBinaryPath(executable, KRB5_BINARY_PATHS);
- }
-
- private ProcessBuilder buildProcessWithKrbEnv(String... argv) throws IOException {
+ private Process startProcessWithKrbEnv(String... argv) throws IOException {
List<String> args = new ArrayList<>();
args.add("env");
for (Map.Entry<String, String> entry : getEnvVars().entrySet()) {
@@ -317,21 +328,36 @@ public class MiniKdc implements Closeable {
}
args.addAll(Arrays.asList(argv));
- LOG.trace("executing {}: {}", Paths.get(argv[0]).getFileName(), Joiner.on(' ').join(args));
+ LOG.debug("executing {}: {}", Paths.get(argv[0]).getFileName(), Joiner.on(' ').join(args));
- return new ProcessBuilder(args).redirectOutput(ProcessBuilder.Redirect.INHERIT)
- .redirectError(ProcessBuilder.Redirect.INHERIT)
- .redirectInput(ProcessBuilder.Redirect.PIPE);
+ return new ProcessBuilder(args).redirectOutput(ProcessBuilder.Redirect.PIPE)
+ .redirectErrorStream(true)
+ .redirectInput(ProcessBuilder.Redirect.PIPE)
+ .start();
}
- private Process startProcessWithKrbEnv(String... argv) throws IOException {
- return buildProcessWithKrbEnv(argv).start();
- }
-
- private static void checkReturnCode(Process process, String name) throws IOException {
+ /**
+ * Waits for the process to exit, checking the return code. Any output to the
+ * process' stdout is optionally logged to SLF4J.
+ * @param process the process to check
+ * @param name the name of the process
+ * @param log whether to log the process' stdout.
+ */
+ private static void checkReturnCode(Process process, String name, boolean log) throws IOException {
int ret;
try {
ret = process.waitFor();
+ if (log) {
+ // Reading the output *after* waiting for the process to close can deadlock
+ // if the process overwhelms the output buffer, however none of the krb5
+ // utilities are known to do that.
+ try (BufferedReader in = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+ String line;
+ while ((line = in.readLine()) != null) {
+ LOG.debug(line);
+ }
+ }
+ }
} catch (InterruptedException e) {
Thread.interrupted();
throw new IOException(String.format("process '%s' interrupted", name));
@@ -341,6 +367,10 @@ public class MiniKdc implements Closeable {
}
}
+ private static String getBinaryPath(String executable) throws IOException {
+ return getBinaryPath(executable, KRB5_BINARY_PATHS);
+ }
+
private static String getBinaryPath(String executable,
List<String> searchPaths) throws IOException {
for (String path : searchPaths) {
@@ -351,7 +381,7 @@ public class MiniKdc implements Closeable {
}
Process which = new ProcessBuilder().command("which", executable).start();
- checkReturnCode(which, "which");
+ checkReturnCode(which, "which", false);
return CharStreams.toString(new InputStreamReader(which.getInputStream())).trim();
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/96c2e2a8/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKuduCluster.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKuduCluster.java b/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKuduCluster.java
index 22e4bfc..37690d9 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKuduCluster.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/MiniKuduCluster.java
@@ -21,6 +21,7 @@ import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.net.HostAndPort;
+
import org.apache.commons.io.FileUtils;
import org.apache.kudu.util.NetUtil;
import org.slf4j.Logger;
@@ -431,7 +432,7 @@ public class MiniKuduCluster implements AutoCloseable {
/**
* Helper runnable that receives stdout and logs it along with the process' identifier.
*/
- private static class ProcessInputStreamLogPrinterRunnable implements Runnable {
+ public static class ProcessInputStreamLogPrinterRunnable implements Runnable {
private final InputStream is;