You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sh...@apache.org on 2018/12/24 17:57:22 UTC
[13/50] hadoop git commit: HDFS-12977. [SBN read] Add stateId to RPC
headers. Contributed by Plamen Jeliazkov.
HDFS-12977. [SBN read] Add stateId to RPC headers. Contributed by Plamen Jeliazkov.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/c9d73437
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/c9d73437
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/c9d73437
Branch: refs/heads/HDFS-12943
Commit: c9d73437e881296282abe46a6b3196befc8cef72
Parents: 3a78e5f
Author: Plamen Jeliazkov <pl...@gmail.com>
Authored: Tue Mar 20 18:48:40 2018 -0700
Committer: Konstantin V Shvachko <sh...@apache.org>
Committed: Mon Dec 24 09:33:59 2018 -0800
----------------------------------------------------------------------
.../org/apache/hadoop/ipc/AlignmentContext.java | 51 ++++++++
.../main/java/org/apache/hadoop/ipc/Client.java | 9 ++
.../apache/hadoop/ipc/ProtobufRpcEngine.java | 9 +-
.../main/java/org/apache/hadoop/ipc/RPC.java | 10 +-
.../java/org/apache/hadoop/ipc/RpcEngine.java | 5 +-
.../main/java/org/apache/hadoop/ipc/Server.java | 15 ++-
.../apache/hadoop/ipc/WritableRpcEngine.java | 33 ++++-
.../src/main/proto/RpcHeader.proto | 1 +
.../java/org/apache/hadoop/ipc/TestRPC.java | 3 +-
.../apache/hadoop/hdfs/ClientGCIContext.java | 65 +++++++++
.../java/org/apache/hadoop/hdfs/DFSClient.java | 3 +
.../server/namenode/GlobalStateIdContext.java | 59 +++++++++
.../hdfs/server/namenode/NameNodeRpcServer.java | 1 +
.../hadoop/hdfs/TestStateAlignmentContext.java | 131 +++++++++++++++++++
14 files changed, 383 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AlignmentContext.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AlignmentContext.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AlignmentContext.java
new file mode 100644
index 0000000..f952325
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AlignmentContext.java
@@ -0,0 +1,51 @@
+/**
+ * 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.hadoop.ipc;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
+
+/**
+ * This interface intends to align the state between client and server
+ * via RPC communication.
+ *
+ * This should be implemented separately on the client side and server side
+ * and can be used to pass state information on RPC responses from server
+ * to client.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Stable
+public interface AlignmentContext {
+
+ /**
+ * This is the intended server method call to implement to pass state info
+ * during RPC response header construction.
+ * @param header The RPC response header builder.
+ */
+ void updateResponseState(RpcResponseHeaderProto.Builder header);
+
+ /**
+ * This is the intended client method call to implement to recieve state info
+ * during RPC response processing.
+ * @param header The RPC response header.
+ */
+ void receiveResponseState(RpcResponseHeaderProto header);
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
index 49f1e49..a33aa1c 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
@@ -103,6 +103,12 @@ public class Client implements AutoCloseable {
return false;
}
};
+ private static AlignmentContext alignmentContext;
+
+ /** Set alignment context to use to fetch state alignment info from RPC. */
+ public static void setAlignmentContext(AlignmentContext ac) {
+ alignmentContext = ac;
+ }
@SuppressWarnings("unchecked")
@Unstable
@@ -1186,6 +1192,9 @@ public class Client implements AutoCloseable {
final Call call = calls.remove(callId);
call.setRpcResponse(value);
}
+ if (alignmentContext != null) {
+ alignmentContext.receiveResponseState(header);
+ }
// verify that packet length was correct
if (packet.remaining() > 0) {
throw new RpcClientException("RPC response length mismatch");
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
index 70fde60..2734a95 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
@@ -337,11 +337,11 @@ public class ProtobufRpcEngine implements RpcEngine {
String bindAddress, int port, int numHandlers, int numReaders,
int queueSizePerHandler, boolean verbose, Configuration conf,
SecretManager<? extends TokenIdentifier> secretManager,
- String portRangeConfig)
+ String portRangeConfig, AlignmentContext alignmentContext)
throws IOException {
return new Server(protocol, protocolImpl, conf, bindAddress, port,
numHandlers, numReaders, queueSizePerHandler, verbose, secretManager,
- portRangeConfig);
+ portRangeConfig, alignmentContext);
}
public static class Server extends RPC.Server {
@@ -410,18 +410,19 @@ public class ProtobufRpcEngine implements RpcEngine {
* @param numHandlers the number of method handler threads to run
* @param verbose whether each call should be logged
* @param portRangeConfig A config parameter that can be used to restrict
- * the range of ports used when port is 0 (an ephemeral port)
+ * @param alignmentContext provides server state info on client responses
*/
public Server(Class<?> protocolClass, Object protocolImpl,
Configuration conf, String bindAddress, int port, int numHandlers,
int numReaders, int queueSizePerHandler, boolean verbose,
SecretManager<? extends TokenIdentifier> secretManager,
- String portRangeConfig)
+ String portRangeConfig, AlignmentContext alignmentContext)
throws IOException {
super(bindAddress, port, null, numHandlers,
numReaders, queueSizePerHandler, conf,
serverNameFromClass(protocolImpl.getClass()), secretManager,
portRangeConfig);
+ setAlignmentContext(alignmentContext);
this.verbose = verbose;
registerProtocolAndImpl(RPC.RpcKind.RPC_PROTOCOL_BUFFER, protocolClass,
protocolImpl);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
index 9cfadc7..36d5400 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
@@ -719,6 +719,7 @@ public class RPC {
private final Configuration conf;
private SecretManager<? extends TokenIdentifier> secretManager = null;
private String portRangeConfig = null;
+ private AlignmentContext alignmentContext = null;
public Builder(Configuration conf) {
this.conf = conf;
@@ -785,6 +786,12 @@ public class RPC {
return this;
}
+ /** Default: null */
+ public Builder setAlignmentContext(AlignmentContext alignmentContext) {
+ this.alignmentContext = alignmentContext;
+ return this;
+ }
+
/**
* Build the RPC Server.
* @throws IOException on error
@@ -804,7 +811,8 @@ public class RPC {
return getProtocolEngine(this.protocol, this.conf).getServer(
this.protocol, this.instance, this.bindAddress, this.port,
this.numHandlers, this.numReaders, this.queueSizePerHandler,
- this.verbose, this.conf, this.secretManager, this.portRangeConfig);
+ this.verbose, this.conf, this.secretManager, this.portRangeConfig,
+ this.alignmentContext);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcEngine.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcEngine.java
index 047722e..8a43172 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcEngine.java
@@ -67,6 +67,7 @@ public interface RpcEngine {
* @param secretManager The secret manager to use to validate incoming requests.
* @param portRangeConfig A config parameter that can be used to restrict
* the range of ports used when port is 0 (an ephemeral port)
+ * @param alignmentContext provides server state info on client responses
* @return The Server instance
* @throws IOException on any error
*/
@@ -75,8 +76,8 @@ public interface RpcEngine {
int queueSizePerHandler, boolean verbose,
Configuration conf,
SecretManager<? extends TokenIdentifier> secretManager,
- String portRangeConfig
- ) throws IOException;
+ String portRangeConfig,
+ AlignmentContext alignmentContext) throws IOException;
/**
* Returns a proxy for ProtocolMetaInfoPB, which uses the given connection
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
index b0ab85c..7705874 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
@@ -140,11 +140,12 @@ public abstract class Server {
private RpcSaslProto negotiateResponse;
private ExceptionsHandler exceptionsHandler = new ExceptionsHandler();
private Tracer tracer;
+ private AlignmentContext alignmentContext;
/**
* Logical name of the server used in metrics and monitor.
*/
private final String serverName;
-
+
/**
* Add exception classes for which server won't log stack traces.
*
@@ -164,6 +165,15 @@ public abstract class Server {
}
/**
+ * Set alignment context to pass state info thru RPC.
+ *
+ * @param alignmentContext alignment state context
+ */
+ public void setAlignmentContext(AlignmentContext alignmentContext) {
+ this.alignmentContext = alignmentContext;
+ }
+
+ /**
* ExceptionsHandler manages Exception groups for special handling
* e.g., terse exception group for concise logging messages
*/
@@ -2977,6 +2987,9 @@ public abstract class Server {
headerBuilder.setRetryCount(call.retryCount);
headerBuilder.setStatus(status);
headerBuilder.setServerIpcVersionNum(CURRENT_VERSION);
+ if(alignmentContext != null) {
+ alignmentContext.updateResponseState(headerBuilder);
+ }
if (status == RpcStatusProto.SUCCESS) {
RpcResponseHeaderProto header = headerBuilder.build();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
index 0497931..507517b 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
@@ -323,11 +323,11 @@ public class WritableRpcEngine implements RpcEngine {
int numHandlers, int numReaders, int queueSizePerHandler,
boolean verbose, Configuration conf,
SecretManager<? extends TokenIdentifier> secretManager,
- String portRangeConfig)
+ String portRangeConfig, AlignmentContext alignmentContext)
throws IOException {
return new Server(protocolClass, protocolImpl, conf, bindAddress, port,
numHandlers, numReaders, queueSizePerHandler, verbose, secretManager,
- portRangeConfig);
+ portRangeConfig, alignmentContext);
}
@@ -397,18 +397,45 @@ public class WritableRpcEngine implements RpcEngine {
* @param port the port to listen for connections on
* @param numHandlers the number of method handler threads to run
* @param verbose whether each call should be logged
+ *
+ * @deprecated use Server#Server(Class, Object,
+ * Configuration, String, int, int, int, int, boolean, SecretManager)
*/
+ @Deprecated
public Server(Class<?> protocolClass, Object protocolImpl,
Configuration conf, String bindAddress, int port,
int numHandlers, int numReaders, int queueSizePerHandler,
boolean verbose, SecretManager<? extends TokenIdentifier> secretManager,
String portRangeConfig)
throws IOException {
+ this(null, protocolImpl, conf, bindAddress, port,
+ numHandlers, numReaders, queueSizePerHandler, verbose,
+ secretManager, null, null);
+ }
+
+ /**
+ * Construct an RPC server.
+ * @param protocolClass - the protocol being registered
+ * can be null for compatibility with old usage (see below for details)
+ * @param protocolImpl the protocol impl that will be called
+ * @param conf the configuration to use
+ * @param bindAddress the address to bind on to listen for connection
+ * @param port the port to listen for connections on
+ * @param numHandlers the number of method handler threads to run
+ * @param verbose whether each call should be logged
+ * @param alignmentContext provides server state info on client responses
+ */
+ public Server(Class<?> protocolClass, Object protocolImpl,
+ Configuration conf, String bindAddress, int port,
+ int numHandlers, int numReaders, int queueSizePerHandler,
+ boolean verbose, SecretManager<? extends TokenIdentifier> secretManager,
+ String portRangeConfig, AlignmentContext alignmentContext)
+ throws IOException {
super(bindAddress, port, null, numHandlers, numReaders,
queueSizePerHandler, conf,
serverNameFromClass(protocolImpl.getClass()), secretManager,
portRangeConfig);
-
+ setAlignmentContext(alignmentContext);
this.verbose = verbose;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/main/proto/RpcHeader.proto
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/proto/RpcHeader.proto b/hadoop-common-project/hadoop-common/src/main/proto/RpcHeader.proto
index aa14616..bfe1301 100644
--- a/hadoop-common-project/hadoop-common/src/main/proto/RpcHeader.proto
+++ b/hadoop-common-project/hadoop-common/src/main/proto/RpcHeader.proto
@@ -155,6 +155,7 @@ message RpcResponseHeaderProto {
optional RpcErrorCodeProto errorDetail = 6; // in case of error
optional bytes clientId = 7; // Globally unique client ID
optional sint32 retryCount = 8 [default = -1];
+ optional int64 stateId = 9; // The last written Global State ID
}
message RpcSaslProto {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java
index c99b403..f8f41ba 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java
@@ -299,7 +299,8 @@ public class TestRPC extends TestRpcBase {
int numHandlers, int numReaders, int queueSizePerHandler,
boolean verbose, Configuration conf,
SecretManager<? extends TokenIdentifier> secretManager,
- String portRangeConfig) throws IOException {
+ String portRangeConfig, AlignmentContext alignmentContext)
+ throws IOException {
return null;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ClientGCIContext.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ClientGCIContext.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ClientGCIContext.java
new file mode 100644
index 0000000..3d722f8
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ClientGCIContext.java
@@ -0,0 +1,65 @@
+/**
+ * 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.hadoop.hdfs;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.ipc.AlignmentContext;
+import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
+
+import java.util.concurrent.atomic.LongAccumulator;
+
+/**
+ * This is the client side implementation responsible for receiving
+ * state alignment info from server(s).
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Stable
+class ClientGCIContext implements AlignmentContext {
+
+ private final DFSClient dfsClient;
+ private final LongAccumulator lastSeenStateId =
+ new LongAccumulator(Math::max, Long.MIN_VALUE);
+
+ /**
+ * Client side constructor.
+ * @param dfsClient client side state receiver
+ */
+ ClientGCIContext(DFSClient dfsClient) {
+ this.dfsClient = dfsClient;
+ }
+
+ /**
+ * Client side implementation only receives state alignment info.
+ * It does not provide state alignment info therefore this does nothing.
+ */
+ @Override
+ public void updateResponseState(RpcResponseHeaderProto.Builder header) {
+ // Do nothing.
+ }
+
+ /**
+ * Client side implementation for receiving state alignment info.
+ */
+ @Override
+ public void receiveResponseState(RpcResponseHeaderProto header) {
+ lastSeenStateId.accumulate(header.getStateId());
+ dfsClient.lastSeenStateId = lastSeenStateId.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index 2badbb1..ecb21c2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -166,6 +166,7 @@ import org.apache.hadoop.io.EnumSetWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.retry.LossyRetryInvocationHandler;
+import org.apache.hadoop.ipc.Client;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.RetriableException;
@@ -219,6 +220,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
final UserGroupInformation ugi;
volatile boolean clientRunning = true;
volatile long lastLeaseRenewal;
+ volatile long lastSeenStateId;
private volatile FsServerDefaults serverDefaults;
private volatile long serverDefaultsLastUpdate;
final String clientName;
@@ -396,6 +398,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
this.saslClient = new SaslDataTransferClient(
conf, DataTransferSaslUtil.getSaslPropertiesResolver(conf),
TrustedChannelResolver.getInstance(conf), nnFallbackToSimpleAuth);
+ Client.setAlignmentContext(new ClientGCIContext(this));
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/GlobalStateIdContext.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/GlobalStateIdContext.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/GlobalStateIdContext.java
new file mode 100644
index 0000000..2d7d94e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/GlobalStateIdContext.java
@@ -0,0 +1,59 @@
+/**
+ * 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.hadoop.hdfs.server.namenode;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.ipc.AlignmentContext;
+import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
+
+/**
+ * This is the server side implementation responsible for passing
+ * state alignment info to clients.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Stable
+class GlobalStateIdContext implements AlignmentContext {
+ private final FSNamesystem namesystem;
+
+ /**
+ * Server side constructor.
+ * @param namesystem server side state provider
+ */
+ GlobalStateIdContext(FSNamesystem namesystem) {
+ this.namesystem = namesystem;
+ }
+
+ /**
+ * Server side implementation for providing state alignment info.
+ */
+ @Override
+ public void updateResponseState(RpcResponseHeaderProto.Builder header) {
+ header.setStateId(namesystem.getLastWrittenTransactionId());
+ }
+
+ /**
+ * Server side implementation only provides state alignment info.
+ * It does not receive state alignment info therefore this does nothing.
+ */
+ @Override
+ public void receiveResponseState(RpcResponseHeaderProto header) {
+ // Do nothing.
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
index f967df4..6ab0c5c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
@@ -456,6 +456,7 @@ public class NameNodeRpcServer implements NamenodeProtocols {
.setNumHandlers(handlerCount)
.setVerbose(false)
.setSecretManager(namesystem.getDelegationTokenSecretManager())
+ .setAlignmentContext(new GlobalStateIdContext(namesystem))
.build();
// Add all the RPC protocols that the namenode implements
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c9d73437/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestStateAlignmentContext.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestStateAlignmentContext.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestStateAlignmentContext.java
new file mode 100644
index 0000000..590f702
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestStateAlignmentContext.java
@@ -0,0 +1,131 @@
+/**
+ * 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.hadoop.hdfs;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+
+/**
+ * Class is used to test server sending state alignment information to clients
+ * via RPC and likewise clients receiving and updating their last known
+ * state alignment info.
+ * These tests check that after a single RPC call a client will have caught up
+ * to the most recent alignment state of the server.
+ */
+public class TestStateAlignmentContext {
+
+ static final long BLOCK_SIZE = 64 * 1024;
+ private static final int NUMDATANODES = 3;
+ private static final Configuration CONF = new HdfsConfiguration();
+
+ private static MiniDFSCluster cluster;
+ private static DistributedFileSystem dfs;
+
+ @BeforeClass
+ public static void startUpCluster() throws IOException {
+ // disable block scanner
+ CONF.setInt(DFSConfigKeys.DFS_DATANODE_SCAN_PERIOD_HOURS_KEY, -1);
+ // Set short retry timeouts so this test runs faster
+ CONF.setInt(HdfsClientConfigKeys.Retry.WINDOW_BASE_KEY, 10);
+ CONF.setBoolean("fs.hdfs.impl.disable.cache", true);
+ cluster = new MiniDFSCluster.Builder(CONF).numDataNodes(NUMDATANODES)
+ .build();
+ cluster.waitActive();
+ }
+
+ @Before
+ public void before() throws IOException {
+ dfs = cluster.getFileSystem();
+ }
+
+ @AfterClass
+ public static void shutDownCluster() throws IOException {
+ if (dfs != null) {
+ dfs.close();
+ dfs = null;
+ }
+ if (cluster != null) {
+ cluster.shutdown();
+ cluster = null;
+ }
+ }
+
+ @After
+ public void after() throws IOException {
+ dfs.close();
+ }
+
+ /**
+ * This test checks if after a client writes we can see the state id in
+ * updated via the response.
+ */
+ @Test
+ public void testStateTransferOnWrite() throws Exception {
+ long preWriteState = cluster.getNamesystem().getLastWrittenTransactionId();
+ DFSTestUtil.writeFile(dfs, new Path("/testFile1"), "abc");
+ long clientState = dfs.dfs.lastSeenStateId;
+ long postWriteState = cluster.getNamesystem().getLastWrittenTransactionId();
+ // Write(s) should have increased state. Check for greater than.
+ assertThat(clientState > preWriteState, is(true));
+ // Client and server state should be equal.
+ assertThat(clientState, is(postWriteState));
+ }
+
+ /**
+ * This test checks if after a client reads we can see the state id in
+ * updated via the response.
+ */
+ @Test
+ public void testStateTransferOnRead() throws Exception {
+ DFSTestUtil.writeFile(dfs, new Path("/testFile2"), "123");
+ long lastWrittenId = cluster.getNamesystem().getLastWrittenTransactionId();
+ DFSTestUtil.readFile(dfs, new Path("/testFile2"));
+ // Read should catch client up to last written state.
+ assertThat(dfs.dfs.lastSeenStateId, is(lastWrittenId));
+ }
+
+ /**
+ * This test checks that a fresh client starts with no state and becomes
+ * updated of state from RPC call.
+ */
+ @Test
+ public void testStateTransferOnFreshClient() throws Exception {
+ DFSTestUtil.writeFile(dfs, new Path("/testFile3"), "ezpz");
+ long lastWrittenId = cluster.getNamesystem().getLastWrittenTransactionId();
+ try (DistributedFileSystem clearDfs =
+ (DistributedFileSystem) FileSystem.get(CONF)) {
+ assertThat(clearDfs.dfs.lastSeenStateId, is(0L));
+ DFSTestUtil.readFile(clearDfs, new Path("/testFile3"));
+ assertThat(clearDfs.dfs.lastSeenStateId, is(lastWrittenId));
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org