szetszwo commented on a change in pull request #2858:
URL: https://github.com/apache/ozone/pull/2858#discussion_r756612127
##########
File path:
hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/ObjectEndpoint.java
##########
@@ -223,13 +233,56 @@ public Response put(
}
}
throw ex;
- } finally {
- if (output != null) {
- output.close();
- }
}
}
+
+ private static void putKeyWithStream(OzoneBucket bucket,
+ String keyPath,
+ long length,
+ int bufferSize,
+ ReplicationConfig replicationConfig,
+ Map<String, String> keyMetadata,
+ InputStream body)
+ throws IOException {
+ try (OzoneDataStreamOutput streamOutput = bucket.createStreamKey(keyPath,
+ length, replicationConfig, keyMetadata)) {
+ writeToStreamOutput(streamOutput, body, bufferSize, length);
+ }
+ }
+
+ private static void writeToStreamOutput(OzoneDataStreamOutput streamOutput,
+ InputStream body,
+ int bufferSize)
+ throws IOException {
+ // (length == -1) Read up to EOF
+ writeToStreamOutput(streamOutput, body, bufferSize, -1);
+ }
+
+ private static void writeToStreamOutput(
+ OzoneDataStreamOutput streamOutput,
+ InputStream body,
+ int bufferSize,
+ long length) throws IOException {
+ byte[] buffer = new byte[bufferSize];
+ ByteBuffer writeByteBuffer;
+ long total = 0;
+ do {
+ int nn = body.read(buffer);
+ if (nn == -1) {
+ break;
+ } else if (nn != bufferSize) {
+ byte[] subBuffer = new byte[nn];
+ System.arraycopy(buffer, 0, subBuffer, 0, nn);
+ writeByteBuffer = ByteBuffer.wrap(subBuffer, 0, nn);
+ } else {
+ writeByteBuffer = ByteBuffer.wrap(buffer, 0, nn);
+ }
+ streamOutput.write(writeByteBuffer, 0, nn);
Review comment:
We may add another put method as below:
```
+++
b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/ObjectEndpoint.java
@@ -155,6 +153,24 @@ public Response put(
@HeaderParam("Content-Length") long length,
@QueryParam("partNumber") int partNumber,
@QueryParam("uploadId") @DefaultValue("") String uploadID,
+ @QueryParam("ozoneInternalParam_streaming") @DefaultValue("false")
+ boolean ozoneInternalParam_streaming,
+ File body) throws IOException, OS3Exception {
+ if (!ozoneInternalParam_streaming) {
+ final BufferedInputStream in = new BufferedInputStream(new
FileInputStream(body));
+ return put(bucketName, keyPath, length, partNumber, uploadID, in);
+ }
+
+ // new code for Ratis streaming
+ ...
+ }
+
+ public Response put(
+ String bucketName,
+ String keyPath,
+ long length,
+ int partNumber,
+ String uploadID,
InputStream body) throws IOException, OS3Exception {
OzoneOutputStream output = null;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
