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/10/23 13:46:39 UTC
hadoop git commit: Revert "HDDS-708. Validate BCSID while reading
blocks from containers in datanodes. Contributed by Shashikant Banerjee."
Repository: hadoop
Updated Branches:
refs/heads/ozone-0.3 a492edee4 -> 575613ef7
Revert "HDDS-708. Validate BCSID while reading blocks from containers in datanodes. Contributed by Shashikant Banerjee."
This reverts commit a492edee4b84e38552f185a4227b49abf60f762d.
The commit msg is wrong.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/575613ef
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/575613ef
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/575613ef
Branch: refs/heads/ozone-0.3
Commit: 575613ef7a051d3a26ef8d895ef10498cb71d667
Parents: a492ede
Author: Shashikant Banerjee <sh...@apache.org>
Authored: Tue Oct 23 19:14:18 2018 +0530
Committer: Shashikant Banerjee <sh...@apache.org>
Committed: Tue Oct 23 19:14:18 2018 +0530
----------------------------------------------------------------------
.../hadoop/hdds/scm/XceiverClientGrpc.java | 194 ++++++-------------
.../hadoop/hdds/scm/XceiverClientManager.java | 14 +-
.../hadoop/hdds/scm/XceiverClientRatis.java | 6 -
.../scm/container/common/helpers/Pipeline.java | 4 -
.../ozone/client/io/ChunkGroupInputStream.java | 9 +-
.../ozone/client/io/ChunkGroupOutputStream.java | 4 -
.../hadoop/ozone/TestMiniOzoneCluster.java | 2 +-
.../ozone/client/rpc/TestOzoneRpcClient.java | 107 ----------
.../ozone/scm/TestXceiverClientManager.java | 15 +-
9 files changed, 70 insertions(+), 285 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
index 9526be3..2f11872 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
@@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.XceiverClientProtocolServiceGrpc;
@@ -41,9 +40,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import java.util.UUID;
-import java.util.Map;
-import java.util.HashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -54,9 +50,9 @@ public class XceiverClientGrpc extends XceiverClientSpi {
static final Logger LOG = LoggerFactory.getLogger(XceiverClientGrpc.class);
private final Pipeline pipeline;
private final Configuration config;
- private Map<UUID, XceiverClientProtocolServiceStub> asyncStubs;
+ private XceiverClientProtocolServiceStub asyncStub;
private XceiverClientMetrics metrics;
- private Map<UUID, ManagedChannel> channels;
+ private ManagedChannel channel;
private final Semaphore semaphore;
private boolean closed = false;
@@ -76,62 +72,46 @@ public class XceiverClientGrpc extends XceiverClientSpi {
this.semaphore =
new Semaphore(HddsClientUtils.getMaxOutstandingRequests(config));
this.metrics = XceiverClientManager.getXceiverClientMetrics();
- this.channels = new HashMap<>();
- this.asyncStubs = new HashMap<>();
}
@Override
public void connect() throws Exception {
-
- // leader by default is the 1st datanode in the datanode list of pipleline
DatanodeDetails leader = this.pipeline.getLeader();
- // just make a connection to the 1st datanode at the beginning
- connectToDatanode(leader);
- }
- private void connectToDatanode(DatanodeDetails dn) {
// read port from the data node, on failure use default configured
// port.
- int port = dn.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue();
+ int port = leader.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue();
if (port == 0) {
port = config.getInt(OzoneConfigKeys.DFS_CONTAINER_IPC_PORT,
OzoneConfigKeys.DFS_CONTAINER_IPC_PORT_DEFAULT);
}
- LOG.debug("Connecting to server Port : " + dn.getIpAddress());
- ManagedChannel channel =
- NettyChannelBuilder.forAddress(dn.getIpAddress(), port).usePlaintext()
- .maxInboundMessageSize(OzoneConfigKeys.DFS_CONTAINER_CHUNK_MAX_SIZE)
- .build();
- XceiverClientProtocolServiceStub asyncStub =
- XceiverClientProtocolServiceGrpc.newStub(channel);
- asyncStubs.put(dn.getUuid(), asyncStub);
- channels.put(dn.getUuid(), channel);
+ LOG.debug("Connecting to server Port : " + leader.getIpAddress());
+ channel = NettyChannelBuilder.forAddress(leader.getIpAddress(), port)
+ .usePlaintext()
+ .maxInboundMessageSize(OzoneConfigKeys.DFS_CONTAINER_CHUNK_MAX_SIZE)
+ .build();
+ asyncStub = XceiverClientProtocolServiceGrpc.newStub(channel);
}
+
/**
- * Returns if the xceiver client connects to all servers in the pipeline.
+ * Returns if the xceiver client connects to a server.
*
* @return True if the connection is alive, false otherwise.
*/
@VisibleForTesting
- public boolean isConnected(DatanodeDetails details) {
- return isConnected(channels.get(details.getUuid()));
- }
-
- private boolean isConnected(ManagedChannel channel) {
- return channel != null && !channel.isTerminated() && !channel.isShutdown();
+ public boolean isConnected() {
+ return !channel.isTerminated() && !channel.isShutdown();
}
@Override
public void close() {
closed = true;
- for (ManagedChannel channel : channels.values()) {
- channel.shutdownNow();
- try {
- channel.awaitTermination(60, TimeUnit.MINUTES);
- } catch (Exception e) {
- LOG.error("Unexpected exception while waiting for channel termination",
- e);
- }
+ channel.shutdownNow();
+ try {
+ channel.awaitTermination(60, TimeUnit.MINUTES);
+ } catch (Exception e) {
+ LOG.error("Unexpected exception while waiting for channel termination",
+ e);
}
}
@@ -140,56 +120,6 @@ public class XceiverClientGrpc extends XceiverClientSpi {
return pipeline;
}
- @Override
- public ContainerCommandResponseProto sendCommand(
- ContainerCommandRequestProto request) throws IOException {
- return sendCommandWithRetry(request);
- }
-
- public ContainerCommandResponseProto sendCommandWithRetry(
- ContainerCommandRequestProto request) throws IOException {
- int size = pipeline.getMachines().size();
- ContainerCommandResponseProto responseProto = null;
- DatanodeDetails dn = null;
-
- // In case of an exception or an error, we will try to read from the
- // datanodes in the pipeline in a round robin fashion.
-
- // TODO: cache the correct leader info in here, so that any subsequent calls
- // should first go to leader
- for (int dnIndex = 0; dnIndex < size; dnIndex++) {
- try {
- dn = pipeline.getMachines().get(dnIndex);
- LOG.debug("Executing command " + request + " on datanode " + dn);
- // In case the command gets retried on a 2nd datanode,
- // sendCommandAsyncCall will create a new channel and async stub
- // in case these don't exist for the specific datanode.
- responseProto = sendCommandAsync(request, dn).get();
- if (responseProto.getResult() == ContainerProtos.Result.SUCCESS) {
- break;
- }
- } catch (ExecutionException | InterruptedException e) {
- LOG.warn("Failed to execute command " + request + " on datanode " + dn
- .getUuidString(), e);
- }
- }
-
- if (responseProto != null) {
- return responseProto;
- } else {
- throw new IOException(
- "Failed to execute command " + request + " on the pipeline "
- + pipeline.getId());
- }
- }
-
- // TODO: for a true async API, once the waitable future while executing
- // the command on one channel fails, it should be retried asynchronously
- // on the future Task for all the remaining datanodes.
-
- // Note: this Async api is not used currently used in any active I/O path.
- // In case it gets used, the asynchronous retry logic needs to be plugged
- // in here.
/**
* Sends a given command to server gets a waitable future back.
*
@@ -198,25 +128,15 @@ public class XceiverClientGrpc extends XceiverClientSpi {
* @throws IOException
*/
@Override
- public CompletableFuture<ContainerCommandResponseProto> sendCommandAsync(
- ContainerCommandRequestProto request)
+ public CompletableFuture<ContainerCommandResponseProto>
+ sendCommandAsync(ContainerCommandRequestProto request)
throws IOException, ExecutionException, InterruptedException {
- return sendCommandAsync(request, pipeline.getLeader());
- }
-
- private CompletableFuture<ContainerCommandResponseProto> sendCommandAsync(
- ContainerCommandRequestProto request, DatanodeDetails dn)
- throws IOException, ExecutionException, InterruptedException {
- if (closed) {
+ if(closed){
throw new IOException("This channel is not connected.");
}
- UUID dnId = dn.getUuid();
- ManagedChannel channel = channels.get(dnId);
- // If the channel doesn't exist for this specific datanode or the channel
- // is closed, just reconnect
- if (!isConnected(channel)) {
- reconnect(dn);
+ if(channel == null || !isConnected()) {
+ reconnect();
}
final CompletableFuture<ContainerCommandResponseProto> replyFuture =
@@ -225,54 +145,48 @@ public class XceiverClientGrpc extends XceiverClientSpi {
long requestTime = Time.monotonicNowNanos();
metrics.incrPendingContainerOpsMetrics(request.getCmdType());
// create a new grpc stream for each non-async call.
-
- // TODO: for async calls, we should reuse StreamObserver resources.
final StreamObserver<ContainerCommandRequestProto> requestObserver =
- asyncStubs.get(dnId)
- .send(new StreamObserver<ContainerCommandResponseProto>() {
- @Override
- public void onNext(ContainerCommandResponseProto value) {
- replyFuture.complete(value);
- metrics.decrPendingContainerOpsMetrics(request.getCmdType());
- metrics.addContainerOpsLatency(request.getCmdType(),
- Time.monotonicNowNanos() - requestTime);
- semaphore.release();
- }
-
- @Override
- public void onError(Throwable t) {
- replyFuture.completeExceptionally(t);
- metrics.decrPendingContainerOpsMetrics(request.getCmdType());
- metrics.addContainerOpsLatency(request.getCmdType(),
- Time.monotonicNowNanos() - requestTime);
- semaphore.release();
- }
-
- @Override
- public void onCompleted() {
- if (!replyFuture.isDone()) {
- replyFuture.completeExceptionally(new IOException(
- "Stream completed but no reply for request " + request));
- }
- }
- });
+ asyncStub.send(new StreamObserver<ContainerCommandResponseProto>() {
+ @Override
+ public void onNext(ContainerCommandResponseProto value) {
+ replyFuture.complete(value);
+ metrics.decrPendingContainerOpsMetrics(request.getCmdType());
+ metrics.addContainerOpsLatency(request.getCmdType(),
+ Time.monotonicNowNanos() - requestTime);
+ semaphore.release();
+ }
+ @Override
+ public void onError(Throwable t) {
+ replyFuture.completeExceptionally(t);
+ metrics.decrPendingContainerOpsMetrics(request.getCmdType());
+ metrics.addContainerOpsLatency(request.getCmdType(),
+ Time.monotonicNowNanos() - requestTime);
+ semaphore.release();
+ }
+
+ @Override
+ public void onCompleted() {
+ if (!replyFuture.isDone()) {
+ replyFuture.completeExceptionally(
+ new IOException("Stream completed but no reply for request "
+ + request));
+ }
+ }
+ });
requestObserver.onNext(request);
requestObserver.onCompleted();
return replyFuture;
}
- private void reconnect(DatanodeDetails dn)
- throws IOException {
- ManagedChannel channel;
+ private void reconnect() throws IOException {
try {
- connectToDatanode(dn);
- channel = channels.get(dn.getUuid());
+ connect();
} catch (Exception e) {
LOG.error("Error while connecting: ", e);
throw new IOException(e);
}
- if (channel == null || !isConnected(channel)) {
+ if (channel == null || !isConnected()) {
throw new IOException("This channel is not connected.");
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java
index 83b5a4c..d542abc 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java
@@ -27,6 +27,7 @@ import com.google.common.cache.RemovalNotification;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.container.common.helpers.PipelineID;
import java.io.Closeable;
import java.io.IOException;
@@ -58,7 +59,7 @@ public class XceiverClientManager implements Closeable {
//TODO : change this to SCM configuration class
private final Configuration conf;
- private final Cache<String, XceiverClientSpi> clientCache;
+ private final Cache<PipelineID, XceiverClientSpi> clientCache;
private final boolean useRatis;
private static XceiverClientMetrics metrics;
@@ -82,10 +83,10 @@ public class XceiverClientManager implements Closeable {
.expireAfterAccess(staleThresholdMs, TimeUnit.MILLISECONDS)
.maximumSize(maxSize)
.removalListener(
- new RemovalListener<String, XceiverClientSpi>() {
+ new RemovalListener<PipelineID, XceiverClientSpi>() {
@Override
public void onRemoval(
- RemovalNotification<String, XceiverClientSpi>
+ RemovalNotification<PipelineID, XceiverClientSpi>
removalNotification) {
synchronized (clientCache) {
// Mark the entry as evicted
@@ -97,7 +98,7 @@ public class XceiverClientManager implements Closeable {
}
@VisibleForTesting
- public Cache<String, XceiverClientSpi> getClientCache() {
+ public Cache<PipelineID, XceiverClientSpi> getClientCache() {
return clientCache;
}
@@ -139,14 +140,13 @@ public class XceiverClientManager implements Closeable {
private XceiverClientSpi getClient(Pipeline pipeline)
throws IOException {
- HddsProtos.ReplicationType type = pipeline.getType();
try {
- return clientCache.get(pipeline.getId().getId().toString() + type,
+ return clientCache.get(pipeline.getId(),
new Callable<XceiverClientSpi>() {
@Override
public XceiverClientSpi call() throws Exception {
XceiverClientSpi client = null;
- switch (type) {
+ switch (pipeline.getType()) {
case RATIS:
client = XceiverClientRatis.newXceiverClientRatis(pipeline, conf);
break;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
index fda30fb..4efe7ba 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdds.scm;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.io.MultipleIOException;
-import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.retry.RetryPolicy;
import org.apache.ratis.thirdparty.com.google.protobuf
.InvalidProtocolBufferException;
@@ -52,7 +51,6 @@ import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
/**
@@ -192,10 +190,6 @@ public final class XceiverClientRatis extends XceiverClientSpi {
getClient().sendAsync(() -> byteString);
}
- public void watchForCommit(long index, long timeout) throws Exception {
- getClient().sendWatchAsync(index, RaftProtos.ReplicationLevel.ALL_COMMITTED)
- .get(timeout, TimeUnit.MILLISECONDS);
- }
/**
* Sends a given command to server gets a waitable future back.
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/Pipeline.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/Pipeline.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/Pipeline.java
index 7fa4c39..49c00a8 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/Pipeline.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/Pipeline.java
@@ -301,10 +301,6 @@ public class Pipeline {
return b.toString();
}
- public void setType(HddsProtos.ReplicationType type) {
- this.type = type;
- }
-
/**
* Returns a JSON string of this object.
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java
index 3772c59..3158334 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java
@@ -22,9 +22,7 @@ import org.apache.hadoop.fs.FSExceptionMessages;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.client.BlockID;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
-import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.hdds.scm.XceiverClientManager;
@@ -278,13 +276,8 @@ public class ChunkGroupInputStream extends InputStream implements Seekable {
long containerID = blockID.getContainerID();
ContainerWithPipeline containerWithPipeline =
storageContainerLocationClient.getContainerWithPipeline(containerID);
- Pipeline pipeline = containerWithPipeline.getPipeline();
-
- // irrespective of the container state, we will always read via Standalone
- // protocol.
- pipeline.setType(HddsProtos.ReplicationType.STAND_ALONE);
XceiverClientSpi xceiverClient = xceiverClientManager
- .acquireClient(pipeline);
+ .acquireClient(containerWithPipeline.getPipeline());
boolean success = false;
containerKey = omKeyLocationInfo.getLocalID();
try {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
index 5966718..0a38a5a 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
@@ -116,10 +116,6 @@ public class ChunkGroupOutputStream extends OutputStream {
public List<ChunkOutputStreamEntry> getStreamEntries() {
return streamEntries;
}
- @VisibleForTesting
- public XceiverClientManager getXceiverClientManager() {
- return xceiverClientManager;
- }
public List<OmKeyLocationInfo> getLocationInfoList() throws IOException {
List<OmKeyLocationInfo> locationInfoList = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java
index bf6a189..302ea46 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java
@@ -102,7 +102,7 @@ public class TestMiniOzoneCluster {
// Verify client is able to connect to the container
try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf)){
client.connect();
- assertTrue(client.isConnected(pipeline.getLeader()));
+ assertTrue(client.isConnected());
}
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java
index 1a401d6..b60343a 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java
@@ -24,10 +24,6 @@ import org.apache.hadoop.hdds.protocol.StorageType;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
-import org.apache.hadoop.hdds.scm.XceiverClientManager;
-import org.apache.hadoop.hdds.scm.XceiverClientRatis;
-import org.apache.hadoop.hdds.scm.XceiverClientSpi;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
import org.apache.hadoop.ozone.*;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -35,7 +31,6 @@ import org.apache.hadoop.ozone.client.*;
import org.apache.hadoop.hdds.client.OzoneQuota;
import org.apache.hadoop.hdds.client.ReplicationFactor;
import org.apache.hadoop.hdds.client.ReplicationType;
-import org.apache.hadoop.ozone.client.io.ChunkGroupOutputStream;
import org.apache.hadoop.ozone.client.io.OzoneInputStream;
import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
@@ -602,108 +597,6 @@ public class TestOzoneRpcClient {
}
@Test
- public void testPutKeyAndGetKeyThreeNodes()
- throws Exception {
- String volumeName = UUID.randomUUID().toString();
- String bucketName = UUID.randomUUID().toString();
-
- String value = "sample value";
- store.createVolume(volumeName);
- OzoneVolume volume = store.getVolume(volumeName);
- volume.createBucket(bucketName);
- OzoneBucket bucket = volume.getBucket(bucketName);
-
- String keyName = UUID.randomUUID().toString();
-
- OzoneOutputStream out = bucket
- .createKey(keyName, value.getBytes().length, ReplicationType.RATIS,
- ReplicationFactor.THREE);
- ChunkGroupOutputStream groupOutputStream =
- (ChunkGroupOutputStream) out.getOutputStream();
- XceiverClientManager manager = groupOutputStream.getXceiverClientManager();
- out.write(value.getBytes());
- out.close();
- // First, confirm the key info from the client matches the info in OM.
- OmKeyArgs.Builder builder = new OmKeyArgs.Builder();
- builder.setVolumeName(volumeName).setBucketName(bucketName)
- .setKeyName(keyName);
- OmKeyLocationInfo keyInfo = ozoneManager.lookupKey(builder.build()).
- getKeyLocationVersions().get(0).getBlocksLatestVersionOnly().get(0);
- long containerID = keyInfo.getContainerID();
- long localID = keyInfo.getLocalID();
- OzoneKeyDetails keyDetails = bucket.getKey(keyName);
- Assert.assertEquals(keyName, keyDetails.getName());
-
- List<OzoneKeyLocation> keyLocations = keyDetails.getOzoneKeyLocations();
- Assert.assertEquals(1, keyLocations.size());
- Assert.assertEquals(containerID, keyLocations.get(0).getContainerID());
- Assert.assertEquals(localID, keyLocations.get(0).getLocalID());
-
- // Make sure that the data size matched.
- Assert
- .assertEquals(value.getBytes().length, keyLocations.get(0).getLength());
-
- ContainerWithPipeline container =
- cluster.getStorageContainerManager().getContainerManager()
- .getContainerWithPipeline(containerID);
- Pipeline pipeline = container.getPipeline();
- List<DatanodeDetails> datanodes = pipeline.getMachines();
-
- DatanodeDetails datanodeDetails = datanodes.get(0);
- Assert.assertNotNull(datanodeDetails);
-
- XceiverClientSpi clientSpi = manager.acquireClient(pipeline);
- Assert.assertTrue(clientSpi instanceof XceiverClientRatis);
- XceiverClientRatis ratisClient = (XceiverClientRatis)clientSpi;
-
- ratisClient.watchForCommit(keyInfo.getBlockCommitSequenceId(), 5000);
- // shutdown the datanode
- cluster.shutdownHddsDatanode(datanodeDetails);
-
- Assert.assertTrue(container.getContainerInfo().getState()
- == HddsProtos.LifeCycleState.OPEN);
- // try to read, this shouls be successful
- readKey(bucket, keyName, value);
-
- Assert.assertTrue(container.getContainerInfo().getState()
- == HddsProtos.LifeCycleState.OPEN);
- // shutdown the second datanode
- datanodeDetails = datanodes.get(1);
- cluster.shutdownHddsDatanode(datanodeDetails);
- Assert.assertTrue(container.getContainerInfo().getState()
- == HddsProtos.LifeCycleState.OPEN);
-
- // the container is open and with loss of 2 nodes we still should be able
- // to read via Standalone protocol
- // try to read
- readKey(bucket, keyName, value);
-
- // shutdown the 3rd datanode
- datanodeDetails = datanodes.get(2);
- cluster.shutdownHddsDatanode(datanodeDetails);
- try {
- // try to read
- readKey(bucket, keyName, value);
- Assert.fail("Expected exception not thrown");
- } catch (IOException e) {
- Assert.assertTrue(e.getMessage().contains("Failed to execute command"));
- Assert.assertTrue(
- e.getMessage().contains("on the pipeline " + pipeline.getId()));
- }
- manager.releaseClient(clientSpi);
- }
-
- private void readKey(OzoneBucket bucket, String keyName, String data)
- throws IOException {
- OzoneKey key = bucket.getKey(keyName);
- Assert.assertEquals(keyName, key.getName());
- OzoneInputStream is = bucket.readKey(keyName);
- byte[] fileContent = new byte[data.getBytes().length];
- is.read(fileContent);
- is.close();
- }
-
- @Test
public void testGetKeyDetails() throws IOException, OzoneException {
String volumeName = UUID.randomUUID().toString();
String bucketName = UUID.randomUUID().toString();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/575613ef/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestXceiverClientManager.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestXceiverClientManager.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestXceiverClientManager.java
index 8b35bbb..da445bf 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestXceiverClientManager.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestXceiverClientManager.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.scm;
import com.google.common.cache.Cache;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.hdds.scm.container.common.helpers.PipelineID;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -106,7 +107,7 @@ public class TestXceiverClientManager {
OzoneConfiguration conf = new OzoneConfiguration();
conf.setInt(SCM_CONTAINER_CLIENT_MAX_SIZE_KEY, 1);
XceiverClientManager clientManager = new XceiverClientManager(conf);
- Cache<String, XceiverClientSpi> cache =
+ Cache<PipelineID, XceiverClientSpi> cache =
clientManager.getClientCache();
ContainerWithPipeline container1 =
@@ -129,9 +130,8 @@ public class TestXceiverClientManager {
Assert.assertNotEquals(client1, client2);
// least recent container (i.e containerName1) is evicted
- XceiverClientSpi nonExistent1 = cache.getIfPresent(
- container1.getContainerInfo().getPipelineID().getId().toString()
- + container1.getContainerInfo().getReplicationType());
+ XceiverClientSpi nonExistent1 = cache
+ .getIfPresent(container1.getContainerInfo().getPipelineID());
Assert.assertEquals(null, nonExistent1);
// However container call should succeed because of refcount on the client.
String traceID1 = "trace" + RandomStringUtils.randomNumeric(4);
@@ -160,7 +160,7 @@ public class TestXceiverClientManager {
OzoneConfiguration conf = new OzoneConfiguration();
conf.setInt(SCM_CONTAINER_CLIENT_MAX_SIZE_KEY, 1);
XceiverClientManager clientManager = new XceiverClientManager(conf);
- Cache<String, XceiverClientSpi> cache =
+ Cache<PipelineID, XceiverClientSpi> cache =
clientManager.getClientCache();
ContainerWithPipeline container1 =
@@ -183,9 +183,8 @@ public class TestXceiverClientManager {
Assert.assertNotEquals(client1, client2);
// now client 1 should be evicted
- XceiverClientSpi nonExistent = cache.getIfPresent(
- container1.getContainerInfo().getPipelineID().getId().toString()
- + container1.getContainerInfo().getReplicationType());
+ XceiverClientSpi nonExistent = cache
+ .getIfPresent(container1.getContainerInfo().getPipelineID());
Assert.assertEquals(null, nonExistent);
// Any container operation should now fail
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org