HDFS-11004. Ozone : move Chunk IO and container protocol calls to hdfs-client. Contributed by Chen Liang.

This commit is contained in:
Anu Engineer 2016-10-13 16:34:29 -07:00 committed by Owen O'Malley
parent ed84388fca
commit ee119ff60a
6 changed files with 115 additions and 94 deletions

View File

@ -29,4 +29,7 @@ public final class ScmConfigKeys {
public static final String DFS_CONTAINER_IPC_PORT = public static final String DFS_CONTAINER_IPC_PORT =
"dfs.container.ipc"; "dfs.container.ipc";
public static final int DFS_CONTAINER_IPC_PORT_DEFAULT = 50011; public static final int DFS_CONTAINER_IPC_PORT_DEFAULT = 50011;
// TODO : this is copied from OzoneConsts, may need to move to a better place
public static final int CHUNK_SIZE = 1 * 1024 * 1024; // 1 MB
} }

View File

@ -16,9 +16,9 @@
* limitations under the License. * limitations under the License.
*/ */
package org.apache.hadoop.ozone.web.storage; package org.apache.hadoop.scm.storage;
import static org.apache.hadoop.ozone.web.storage.ContainerProtocolCalls.*; import static org.apache.hadoop.scm.storage.ContainerProtocolCalls.*;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
@ -31,23 +31,21 @@
import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo; import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo;
import org.apache.hadoop.scm.XceiverClient; import org.apache.hadoop.scm.XceiverClient;
import org.apache.hadoop.scm.XceiverClientManager; import org.apache.hadoop.scm.XceiverClientManager;
import org.apache.hadoop.ozone.web.exceptions.OzoneException;
import org.apache.hadoop.ozone.web.handlers.UserArgs;
/** /**
* An {@link InputStream} used by the REST service in combination with the * An {@link InputStream} used by the REST service in combination with the
* {@link DistributedStorageHandler} to read the value of a key from a sequence * SCMClient to read the value of a key from a sequence
* of container chunks. All bytes of the key value are stored in container * of container chunks. All bytes of the key value are stored in container
* chunks. Each chunk may contain multiple underlying {@link ByteBuffer} * chunks. Each chunk may contain multiple underlying {@link ByteBuffer}
* instances. This class encapsulates all state management for iterating * instances. This class encapsulates all state management for iterating
* through the sequence of chunks and the sequence of buffers within each chunk. * through the sequence of chunks and the sequence of buffers within each chunk.
*/ */
class ChunkInputStream extends InputStream { public class ChunkInputStream extends InputStream {
private static final int EOF = -1; private static final int EOF = -1;
private final String key; private final String key;
private final UserArgs args; private final String traceID;
private XceiverClientManager xceiverClientManager; private XceiverClientManager xceiverClientManager;
private XceiverClient xceiverClient; private XceiverClient xceiverClient;
private List<ChunkInfo> chunks; private List<ChunkInfo> chunks;
@ -62,12 +60,12 @@ class ChunkInputStream extends InputStream {
* @param xceiverClientManager client manager that controls client * @param xceiverClientManager client manager that controls client
* @param xceiverClient client to perform container calls * @param xceiverClient client to perform container calls
* @param chunks list of chunks to read * @param chunks list of chunks to read
* @param args container protocol call args * @param traceID container protocol call traceID
*/ */
public ChunkInputStream(String key, XceiverClientManager xceiverClientManager, public ChunkInputStream(String key, XceiverClientManager xceiverClientManager,
XceiverClient xceiverClient, List<ChunkInfo> chunks, UserArgs args) { XceiverClient xceiverClient, List<ChunkInfo> chunks, String traceID) {
this.key = key; this.key = key;
this.args = args; this.traceID = traceID;
this.xceiverClientManager = xceiverClientManager; this.xceiverClientManager = xceiverClientManager;
this.xceiverClient = xceiverClient; this.xceiverClient = xceiverClient;
this.chunks = chunks; this.chunks = chunks;
@ -182,8 +180,8 @@ private synchronized void readChunkFromContainer(int readChunkOffset)
final ReadChunkResponseProto readChunkResponse; final ReadChunkResponseProto readChunkResponse;
try { try {
readChunkResponse = readChunk(xceiverClient, chunks.get(readChunkOffset), readChunkResponse = readChunk(xceiverClient, chunks.get(readChunkOffset),
key, args); key, traceID);
} catch (OzoneException e) { } catch (IOException e) {
throw new IOException("Unexpected OzoneException", e); throw new IOException("Unexpected OzoneException", e);
} }
chunkOffset = readChunkOffset; chunkOffset = readChunkOffset;

View File

@ -16,11 +16,10 @@
* limitations under the License. * limitations under the License.
*/ */
package org.apache.hadoop.ozone.web.storage; package org.apache.hadoop.scm.storage;
import static org.apache.hadoop.ozone.OzoneConsts.CHUNK_SIZE; import static org.apache.hadoop.scm.storage.ContainerProtocolCalls.putKey;
import static org.apache.hadoop.ozone.web.storage.ContainerProtocolCalls.*; import static org.apache.hadoop.scm.storage.ContainerProtocolCalls.writeChunk;
import static org.apache.hadoop.ozone.web.storage.OzoneContainerTranslation.*;
import java.io.IOException; import java.io.IOException;
import java.io.OutputStream; import java.io.OutputStream;
@ -31,15 +30,14 @@
import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo; import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo;
import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyData; import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyData;
import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyValue;
import org.apache.hadoop.scm.ScmConfigKeys;
import org.apache.hadoop.scm.XceiverClient; import org.apache.hadoop.scm.XceiverClient;
import org.apache.hadoop.scm.XceiverClientManager; import org.apache.hadoop.scm.XceiverClientManager;
import org.apache.hadoop.ozone.web.exceptions.OzoneException;
import org.apache.hadoop.ozone.web.handlers.UserArgs;
import org.apache.hadoop.ozone.web.response.KeyInfo;
/** /**
* An {@link OutputStream} used by the REST service in combination with the * An {@link OutputStream} used by the REST service in combination with the
* {@link DistributedStorageHandler} to write the value of a key to a sequence * SCMClient to write the value of a key to a sequence
* of container chunks. Writes are buffered locally and periodically written to * of container chunks. Writes are buffered locally and periodically written to
* the container as a new chunk. In order to preserve the semantics that * the container as a new chunk. In order to preserve the semantics that
* replacement of a pre-existing key is atomic, each instance of the stream has * replacement of a pre-existing key is atomic, each instance of the stream has
@ -53,11 +51,11 @@
* This class encapsulates all state management for buffering and writing * This class encapsulates all state management for buffering and writing
* through to the container. * through to the container.
*/ */
class ChunkOutputStream extends OutputStream { public class ChunkOutputStream extends OutputStream {
private final String containerKey; private final String containerKey;
private final KeyInfo key; private final String key;
private final UserArgs args; private final String traceID;
private final KeyData.Builder containerKeyData; private final KeyData.Builder containerKeyData;
private XceiverClientManager xceiverClientManager; private XceiverClientManager xceiverClientManager;
private XceiverClient xceiverClient; private XceiverClient xceiverClient;
@ -72,19 +70,23 @@ class ChunkOutputStream extends OutputStream {
* @param key chunk key * @param key chunk key
* @param xceiverClientManager client manager that controls client * @param xceiverClientManager client manager that controls client
* @param xceiverClient client to perform container calls * @param xceiverClient client to perform container calls
* @param args container protocol call args * @param traceID container protocol call args
*/ */
public ChunkOutputStream(String containerKey, KeyInfo key, public ChunkOutputStream(String containerKey, String key,
XceiverClientManager xceiverClientManager, XceiverClient xceiverClient, XceiverClientManager xceiverClientManager, XceiverClient xceiverClient,
UserArgs args) { String traceID) {
this.containerKey = containerKey; this.containerKey = containerKey;
this.key = key; this.key = key;
this.args = args; this.traceID = traceID;
this.containerKeyData = fromKeyToContainerKeyDataBuilder( KeyValue keyValue = KeyValue.newBuilder()
xceiverClient.getPipeline().getContainerName(), containerKey, key); .setKey("TYPE").setValue("KEY").build();
this.containerKeyData = KeyData.newBuilder()
.setContainerName(xceiverClient.getPipeline().getContainerName())
.setName(containerKey)
.addMetadata(keyValue);
this.xceiverClientManager = xceiverClientManager; this.xceiverClientManager = xceiverClientManager;
this.xceiverClient = xceiverClient; this.xceiverClient = xceiverClient;
this.buffer = ByteBuffer.allocate(CHUNK_SIZE); this.buffer = ByteBuffer.allocate(ScmConfigKeys.CHUNK_SIZE);
this.streamId = UUID.randomUUID().toString(); this.streamId = UUID.randomUUID().toString();
this.chunkIndex = 0; this.chunkIndex = 0;
} }
@ -95,7 +97,7 @@ public synchronized void write(int b) throws IOException {
int rollbackPosition = buffer.position(); int rollbackPosition = buffer.position();
int rollbackLimit = buffer.limit(); int rollbackLimit = buffer.limit();
buffer.put((byte)b); buffer.put((byte)b);
if (buffer.position() == CHUNK_SIZE) { if (buffer.position() == ScmConfigKeys.CHUNK_SIZE) {
flushBufferToChunk(rollbackPosition, rollbackLimit); flushBufferToChunk(rollbackPosition, rollbackLimit);
} }
} }
@ -114,11 +116,12 @@ public void write(byte[] b, int off, int len) throws IOException {
} }
checkOpen(); checkOpen();
while (len > 0) { while (len > 0) {
int writeLen = Math.min(CHUNK_SIZE - buffer.position(), len); int writeLen = Math.min(
ScmConfigKeys.CHUNK_SIZE - buffer.position(), len);
int rollbackPosition = buffer.position(); int rollbackPosition = buffer.position();
int rollbackLimit = buffer.limit(); int rollbackLimit = buffer.limit();
buffer.put(b, off, writeLen); buffer.put(b, off, writeLen);
if (buffer.position() == CHUNK_SIZE) { if (buffer.position() == ScmConfigKeys.CHUNK_SIZE) {
flushBufferToChunk(rollbackPosition, rollbackLimit); flushBufferToChunk(rollbackPosition, rollbackLimit);
} }
off += writeLen; off += writeLen;
@ -144,9 +147,9 @@ public synchronized void close() throws IOException {
if (buffer.position() > 0) { if (buffer.position() > 0) {
writeChunkToContainer(); writeChunkToContainer();
} }
putKey(xceiverClient, containerKeyData.build(), args); putKey(xceiverClient, containerKeyData.build(), traceID);
} catch (OzoneException e) { } catch (IOException e) {
throw new IOException("Unexpected OzoneException", e); throw new IOException("Unexpected Storage Container Exception", e);
} finally { } finally {
xceiverClientManager.releaseClient(xceiverClient); xceiverClientManager.releaseClient(xceiverClient);
xceiverClientManager = null; xceiverClientManager = null;
@ -205,14 +208,14 @@ private synchronized void writeChunkToContainer() throws IOException {
ChunkInfo chunk = ChunkInfo ChunkInfo chunk = ChunkInfo
.newBuilder() .newBuilder()
.setChunkName( .setChunkName(
key.getKeyName() + "_stream_" + streamId + "_chunk_" + ++chunkIndex) key + "_stream_" + streamId + "_chunk_" + ++chunkIndex)
.setOffset(0) .setOffset(0)
.setLen(data.size()) .setLen(data.size())
.build(); .build();
try { try {
writeChunk(xceiverClient, chunk, key.getKeyName(), data, args); writeChunk(xceiverClient, chunk, key, data, traceID);
} catch (OzoneException e) { } catch (IOException e) {
throw new IOException("Unexpected OzoneException", e); throw new IOException("Unexpected Storage Container Exception", e);
} }
containerKeyData.addChunks(chunk); containerKeyData.addChunks(chunk);
} }

View File

@ -16,7 +16,7 @@
* limitations under the License. * limitations under the License.
*/ */
package org.apache.hadoop.ozone.web.storage; package org.apache.hadoop.scm.storage;
import static java.net.HttpURLConnection.HTTP_BAD_REQUEST; import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR; import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
@ -37,29 +37,24 @@
import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Type; import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Type;
import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.WriteChunkRequestProto; import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.WriteChunkRequestProto;
import org.apache.hadoop.scm.XceiverClient; import org.apache.hadoop.scm.XceiverClient;
import org.apache.hadoop.ozone.web.exceptions.ErrorTable;
import org.apache.hadoop.ozone.web.exceptions.OzoneException;
import org.apache.hadoop.ozone.web.handlers.UserArgs;
/** /**
* Implementation of all container protocol calls performed by * Implementation of all container protocol calls performed by
* {@link DistributedStorageHandler}. * .
*/ */
final class ContainerProtocolCalls { public final class ContainerProtocolCalls {
/** /**
* Calls the container protocol to get a container key. * Calls the container protocol to get a container key.
* *
* @param xceiverClient client to perform call * @param xceiverClient client to perform call
* @param containerKeyData key data to identify container * @param containerKeyData key data to identify container
* @param args container protocol call args * @param traceID container protocol call args
* @returns container protocol get key response * @return container protocol get key response
* @throws IOException if there is an I/O error while performing the call * @throws IOException if there is an I/O error while performing the call
* @throws OzoneException if the container protocol call failed
*/ */
public static GetKeyResponseProto getKey(XceiverClient xceiverClient, public static GetKeyResponseProto getKey(XceiverClient xceiverClient,
KeyData containerKeyData, UserArgs args) throws IOException, KeyData containerKeyData, String traceID) throws IOException {
OzoneException {
GetKeyRequestProto.Builder readKeyRequest = GetKeyRequestProto GetKeyRequestProto.Builder readKeyRequest = GetKeyRequestProto
.newBuilder() .newBuilder()
.setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
@ -67,11 +62,11 @@ public static GetKeyResponseProto getKey(XceiverClient xceiverClient,
ContainerCommandRequestProto request = ContainerCommandRequestProto ContainerCommandRequestProto request = ContainerCommandRequestProto
.newBuilder() .newBuilder()
.setCmdType(Type.GetKey) .setCmdType(Type.GetKey)
.setTraceID(args.getRequestID()) .setTraceID(traceID)
.setGetKey(readKeyRequest) .setGetKey(readKeyRequest)
.build(); .build();
ContainerCommandResponseProto response = xceiverClient.sendCommand(request); ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
validateContainerResponse(response, args); validateContainerResponse(response, traceID);
return response.getGetKey(); return response.getGetKey();
} }
@ -80,13 +75,11 @@ public static GetKeyResponseProto getKey(XceiverClient xceiverClient,
* *
* @param xceiverClient client to perform call * @param xceiverClient client to perform call
* @param containerKeyData key data to identify container * @param containerKeyData key data to identify container
* @param args container protocol call args * @param traceID container protocol call args
* @throws IOException if there is an I/O error while performing the call * @throws IOException if there is an I/O error while performing the call
* @throws OzoneException if the container protocol call failed
*/ */
public static void putKey(XceiverClient xceiverClient, public static void putKey(XceiverClient xceiverClient,
KeyData containerKeyData, UserArgs args) throws IOException, KeyData containerKeyData, String traceID) throws IOException {
OzoneException {
PutKeyRequestProto.Builder createKeyRequest = PutKeyRequestProto PutKeyRequestProto.Builder createKeyRequest = PutKeyRequestProto
.newBuilder() .newBuilder()
.setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
@ -94,11 +87,11 @@ public static void putKey(XceiverClient xceiverClient,
ContainerCommandRequestProto request = ContainerCommandRequestProto ContainerCommandRequestProto request = ContainerCommandRequestProto
.newBuilder() .newBuilder()
.setCmdType(Type.PutKey) .setCmdType(Type.PutKey)
.setTraceID(args.getRequestID()) .setTraceID(traceID)
.setPutKey(createKeyRequest) .setPutKey(createKeyRequest)
.build(); .build();
ContainerCommandResponseProto response = xceiverClient.sendCommand(request); ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
validateContainerResponse(response, args); validateContainerResponse(response, traceID);
} }
/** /**
@ -107,14 +100,13 @@ public static void putKey(XceiverClient xceiverClient,
* @param xceiverClient client to perform call * @param xceiverClient client to perform call
* @param chunk information about chunk to read * @param chunk information about chunk to read
* @param key the key name * @param key the key name
* @param args container protocol call args * @param traceID container protocol call args
* @returns container protocol read chunk response * @return container protocol read chunk response
* @throws IOException if there is an I/O error while performing the call * @throws IOException if there is an I/O error while performing the call
* @throws OzoneException if the container protocol call failed
*/ */
public static ReadChunkResponseProto readChunk(XceiverClient xceiverClient, public static ReadChunkResponseProto readChunk(XceiverClient xceiverClient,
ChunkInfo chunk, String key, UserArgs args) ChunkInfo chunk, String key, String traceID)
throws IOException, OzoneException { throws IOException {
ReadChunkRequestProto.Builder readChunkRequest = ReadChunkRequestProto ReadChunkRequestProto.Builder readChunkRequest = ReadChunkRequestProto
.newBuilder() .newBuilder()
.setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
@ -123,11 +115,11 @@ public static ReadChunkResponseProto readChunk(XceiverClient xceiverClient,
ContainerCommandRequestProto request = ContainerCommandRequestProto ContainerCommandRequestProto request = ContainerCommandRequestProto
.newBuilder() .newBuilder()
.setCmdType(Type.ReadChunk) .setCmdType(Type.ReadChunk)
.setTraceID(args.getRequestID()) .setTraceID(traceID)
.setReadChunk(readChunkRequest) .setReadChunk(readChunkRequest)
.build(); .build();
ContainerCommandResponseProto response = xceiverClient.sendCommand(request); ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
validateContainerResponse(response, args); validateContainerResponse(response, traceID);
return response.getReadChunk(); return response.getReadChunk();
} }
@ -138,13 +130,12 @@ public static ReadChunkResponseProto readChunk(XceiverClient xceiverClient,
* @param chunk information about chunk to write * @param chunk information about chunk to write
* @param key the key name * @param key the key name
* @param data the data of the chunk to write * @param data the data of the chunk to write
* @param args container protocol call args * @param traceID container protocol call args
* @throws IOException if there is an I/O error while performing the call * @throws IOException if there is an I/O error while performing the call
* @throws OzoneException if the container protocol call failed
*/ */
public static void writeChunk(XceiverClient xceiverClient, ChunkInfo chunk, public static void writeChunk(XceiverClient xceiverClient, ChunkInfo chunk,
String key, ByteString data, UserArgs args) String key, ByteString data, String traceID)
throws IOException, OzoneException { throws IOException {
WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto
.newBuilder() .newBuilder()
.setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
@ -154,11 +145,11 @@ public static void writeChunk(XceiverClient xceiverClient, ChunkInfo chunk,
ContainerCommandRequestProto request = ContainerCommandRequestProto ContainerCommandRequestProto request = ContainerCommandRequestProto
.newBuilder() .newBuilder()
.setCmdType(Type.WriteChunk) .setCmdType(Type.WriteChunk)
.setTraceID(args.getRequestID()) .setTraceID(traceID)
.setWriteChunk(writeChunkRequest) .setWriteChunk(writeChunkRequest)
.build(); .build();
ContainerCommandResponseProto response = xceiverClient.sendCommand(request); ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
validateContainerResponse(response, args); validateContainerResponse(response, traceID);
} }
/** /**
@ -166,27 +157,28 @@ public static void writeChunk(XceiverClient xceiverClient, ChunkInfo chunk,
* return code is mapped to a corresponding exception and thrown. * return code is mapped to a corresponding exception and thrown.
* *
* @param response container protocol call response * @param response container protocol call response
* @param args container protocol call args * @param traceID container protocol call args
* @throws OzoneException if the container protocol call failed * @throws IOException if the container protocol call failed
*/ */
private static void validateContainerResponse( private static void validateContainerResponse(
ContainerCommandResponseProto response, UserArgs args) ContainerCommandResponseProto response, String traceID
throws OzoneException { ) throws IOException {
// TODO : throw the right type of exception
switch (response.getResult()) { switch (response.getResult()) {
case SUCCESS: case SUCCESS:
break; break;
case MALFORMED_REQUEST: case MALFORMED_REQUEST:
throw ErrorTable.newError(new OzoneException(HTTP_BAD_REQUEST, throw new IOException(HTTP_BAD_REQUEST +
"badRequest", "Bad container request."), args); ":Bad container request: " + traceID);
case UNSUPPORTED_REQUEST: case UNSUPPORTED_REQUEST:
throw ErrorTable.newError(new OzoneException(HTTP_INTERNAL_ERROR, throw new IOException(HTTP_INTERNAL_ERROR +
"internalServerError", "Unsupported container request."), args); "Unsupported container request: " + traceID);
case CONTAINER_INTERNAL_ERROR: case CONTAINER_INTERNAL_ERROR:
throw ErrorTable.newError(new OzoneException(HTTP_INTERNAL_ERROR, throw new IOException(HTTP_INTERNAL_ERROR +
"internalServerError", "Container internal error."), args); "Container internal error:" + traceID);
default: default:
throw ErrorTable.newError(new OzoneException(HTTP_INTERNAL_ERROR, throw new IOException(HTTP_INTERNAL_ERROR +
"internalServerError", "Unrecognized container response."), args); "Unrecognized container response:" + traceID);
} }
} }

View File

@ -0,0 +1,23 @@
/**
* 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.scm.storage;
/**
* This package contains StorageContainerManager classes.
*/

View File

@ -18,7 +18,7 @@
package org.apache.hadoop.ozone.web.storage; package org.apache.hadoop.ozone.web.storage;
import static org.apache.hadoop.ozone.web.storage.ContainerProtocolCalls.*; import static org.apache.hadoop.scm.storage.ContainerProtocolCalls.*;
import static org.apache.hadoop.ozone.web.storage.OzoneContainerTranslation.*; import static org.apache.hadoop.ozone.web.storage.OzoneContainerTranslation.*;
import java.io.IOException; import java.io.IOException;
@ -57,6 +57,8 @@
import org.apache.hadoop.ozone.web.response.ListVolumes; import org.apache.hadoop.ozone.web.response.ListVolumes;
import org.apache.hadoop.ozone.web.response.VolumeInfo; import org.apache.hadoop.ozone.web.response.VolumeInfo;
import org.apache.hadoop.ozone.web.response.VolumeOwner; import org.apache.hadoop.ozone.web.response.VolumeOwner;
import org.apache.hadoop.scm.storage.ChunkInputStream;
import org.apache.hadoop.scm.storage.ChunkOutputStream;
import org.apache.hadoop.util.StringUtils; import org.apache.hadoop.util.StringUtils;
/** /**
@ -95,7 +97,7 @@ public void createVolume(VolumeArgs args) throws IOException, OzoneException {
volume.setCreatedBy(args.getAdminName()); volume.setCreatedBy(args.getAdminName());
KeyData containerKeyData = fromVolumeToContainerKeyData( KeyData containerKeyData = fromVolumeToContainerKeyData(
xceiverClient.getPipeline().getContainerName(), containerKey, volume); xceiverClient.getPipeline().getContainerName(), containerKey, volume);
putKey(xceiverClient, containerKeyData, args); putKey(xceiverClient, containerKeyData, args.getRequestID());
} finally { } finally {
xceiverClientManager.releaseClient(xceiverClient); xceiverClientManager.releaseClient(xceiverClient);
} }
@ -140,7 +142,7 @@ public VolumeInfo getVolumeInfo(VolumeArgs args)
KeyData containerKeyData = containerKeyDataForRead( KeyData containerKeyData = containerKeyDataForRead(
xceiverClient.getPipeline().getContainerName(), containerKey); xceiverClient.getPipeline().getContainerName(), containerKey);
GetKeyResponseProto response = getKey(xceiverClient, containerKeyData, GetKeyResponseProto response = getKey(xceiverClient, containerKeyData,
args); args.getRequestID());
return fromContainerKeyValueListToVolume( return fromContainerKeyValueListToVolume(
response.getKeyData().getMetadataList()); response.getKeyData().getMetadataList());
} finally { } finally {
@ -163,7 +165,7 @@ public void createBucket(final BucketArgs args)
bucket.setStorageType(args.getStorageType()); bucket.setStorageType(args.getStorageType());
KeyData containerKeyData = fromBucketToContainerKeyData( KeyData containerKeyData = fromBucketToContainerKeyData(
xceiverClient.getPipeline().getContainerName(), containerKey, bucket); xceiverClient.getPipeline().getContainerName(), containerKey, bucket);
putKey(xceiverClient, containerKeyData, args); putKey(xceiverClient, containerKeyData, args.getRequestID());
} finally { } finally {
xceiverClientManager.releaseClient(xceiverClient); xceiverClientManager.releaseClient(xceiverClient);
} }
@ -218,7 +220,7 @@ public BucketInfo getBucketInfo(BucketArgs args)
KeyData containerKeyData = containerKeyDataForRead( KeyData containerKeyData = containerKeyDataForRead(
xceiverClient.getPipeline().getContainerName(), containerKey); xceiverClient.getPipeline().getContainerName(), containerKey);
GetKeyResponseProto response = getKey(xceiverClient, containerKeyData, GetKeyResponseProto response = getKey(xceiverClient, containerKeyData,
args); args.getRequestID());
return fromContainerKeyValueListToBucket( return fromContainerKeyValueListToBucket(
response.getKeyData().getMetadataList()); response.getKeyData().getMetadataList());
} finally { } finally {
@ -235,8 +237,8 @@ public OutputStream newKeyWriter(KeyArgs args) throws IOException,
key.setKeyName(args.getKeyName()); key.setKeyName(args.getKeyName());
key.setCreatedOn(dateToString(new Date())); key.setCreatedOn(dateToString(new Date()));
XceiverClient xceiverClient = acquireXceiverClient(containerKey); XceiverClient xceiverClient = acquireXceiverClient(containerKey);
return new ChunkOutputStream(containerKey, key, xceiverClientManager, return new ChunkOutputStream(containerKey, key.getKeyName(),
xceiverClient, args); xceiverClientManager, xceiverClient, args.getRequestID());
} }
@Override @Override
@ -256,7 +258,7 @@ public LengthInputStream newKeyReader(KeyArgs args) throws IOException,
KeyData containerKeyData = containerKeyDataForRead( KeyData containerKeyData = containerKeyDataForRead(
xceiverClient.getPipeline().getContainerName(), containerKey); xceiverClient.getPipeline().getContainerName(), containerKey);
GetKeyResponseProto response = getKey(xceiverClient, containerKeyData, GetKeyResponseProto response = getKey(xceiverClient, containerKeyData,
args); args.getRequestID());
long length = 0; long length = 0;
List<ChunkInfo> chunks = response.getKeyData().getChunksList(); List<ChunkInfo> chunks = response.getKeyData().getChunksList();
for (ChunkInfo chunk : chunks) { for (ChunkInfo chunk : chunks) {
@ -264,8 +266,8 @@ public LengthInputStream newKeyReader(KeyArgs args) throws IOException,
} }
success = true; success = true;
return new LengthInputStream(new ChunkInputStream( return new LengthInputStream(new ChunkInputStream(
containerKey, xceiverClientManager, xceiverClient, chunks, args), containerKey, xceiverClientManager, xceiverClient,
length); chunks, args.getRequestID()), length);
} finally { } finally {
if (!success) { if (!success) {
xceiverClientManager.releaseClient(xceiverClient); xceiverClientManager.releaseClient(xceiverClient);