Skip to content

Commit f646b1e

Browse files
peterxclipeterxcli
authored
HDDS-15213. Use common commit output for stream outputs (#10224)
1 parent 7b82918 commit f646b1e

7 files changed

Lines changed: 153 additions & 28 deletions

File tree

‎hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyCommitOutput.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.hadoop.ozone.client.io;
19+
20+
import jakarta.annotation.Nonnull;
21+
import java.io.IOException;
22+
import java.util.List;
23+
import org.apache.hadoop.ozone.om.helpers.OmMultipartCommitUploadPartInfo;
24+
import org.apache.ratis.util.function.CheckedRunnable;
25+
26+
/**
27+
* Common commit-time behavior for key output implementations.
28+
*/
29+
interface KeyCommitOutput extends KeyMetadataAware {
30+
31+
void setPreCommits(
32+
@Nonnull List<CheckedRunnable<IOException>> preCommits);
33+
34+
OmMultipartCommitUploadPartInfo getCommitUploadPartInfo();
35+
}

‎hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@
6060
* TODO : currently not support multi-thread access.
6161
*/
6262
public class KeyDataStreamOutput extends AbstractDataStreamOutput
63-
implements KeyMetadataAware {
63+
implements KeyCommitOutput {
6464

6565
private static final Logger LOG =
6666
LoggerFactory.getLogger(KeyDataStreamOutput.class);
@@ -87,6 +87,7 @@ public class KeyDataStreamOutput extends AbstractDataStreamOutput
8787

8888
private List<CheckedRunnable<IOException>> preCommits = Collections.emptyList();
8989

90+
@Override
9091
public void setPreCommits(@Nonnull List<CheckedRunnable<IOException>> preCommits) {
9192
this.preCommits = preCommits;
9293
}
@@ -472,6 +473,7 @@ public void close() throws IOException {
472473
}
473474
}
474475

476+
@Override
475477
public OmMultipartCommitUploadPartInfo getCommitUploadPartInfo() {
476478
return blockDataStreamOutputEntryPool.getCommitUploadPartInfo();
477479
}

‎hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyOutputStream.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@
7676
* TODO : currently not support multi-thread access.
7777
*/
7878
public class KeyOutputStream extends OutputStream
79-
implements Syncable, KeyMetadataAware {
79+
implements Syncable, KeyCommitOutput {
8080

8181
private static final Logger LOG =
8282
LoggerFactory.getLogger(KeyOutputStream.class);
@@ -114,6 +114,7 @@ public class KeyOutputStream extends OutputStream
114114
private final KeyOutputStreamSemaphore keyOutputStreamSemaphore;
115115
private List<CheckedRunnable<IOException>> preCommits = Collections.emptyList();
116116

117+
@Override
117118
public void setPreCommits(@Nonnull List<CheckedRunnable<IOException>> preCommits) {
118119
this.preCommits = preCommits;
119120
}
@@ -671,7 +672,8 @@ private void closeInternal() throws IOException {
671672
}
672673
}
673674

674-
synchronized OmMultipartCommitUploadPartInfo
675+
@Override
676+
public synchronized OmMultipartCommitUploadPartInfo
675677
getCommitUploadPartInfo() {
676678
return blockOutputStreamEntryPool.getCommitUploadPartInfo();
677679
}

‎hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneDataStreamOutput.java‎

Lines changed: 48 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,15 @@
2020
import java.io.IOException;
2121
import java.io.OutputStream;
2222
import java.nio.ByteBuffer;
23+
import java.util.List;
2324
import java.util.Map;
2425
import java.util.Objects;
2526
import java.util.Optional;
2627
import org.apache.hadoop.crypto.CryptoOutputStream;
2728
import org.apache.hadoop.fs.Syncable;
2829
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
2930
import org.apache.hadoop.ozone.om.helpers.OmMultipartCommitUploadPartInfo;
31+
import org.apache.ratis.util.function.CheckedRunnable;
3032

3133
/**
3234
* OzoneDataStreamOutput is used to write data into Ozone.
@@ -100,40 +102,67 @@ public synchronized void close() throws IOException {
100102
}
101103

102104
public OmMultipartCommitUploadPartInfo getCommitUploadPartInfo() {
103-
KeyDataStreamOutput keyDataStreamOutput = getKeyDataStreamOutput();
104-
if (keyDataStreamOutput != null) {
105-
return keyDataStreamOutput.getCommitUploadPartInfo();
105+
KeyCommitOutput keyCommitOutput = getKeyCommitOutput();
106+
if (keyCommitOutput != null) {
107+
return keyCommitOutput.getCommitUploadPartInfo();
106108
}
107109
// Otherwise return null.
108110
return null;
109111
}
110112

111113
public KeyDataStreamOutput getKeyDataStreamOutput() {
114+
if (byteBufferStreamOutput instanceof KeyDataStreamOutput) {
115+
return ((KeyDataStreamOutput) byteBufferStreamOutput);
116+
}
112117
if (byteBufferStreamOutput instanceof OzoneOutputStream) {
113118
OutputStream outputStream =
114119
((OzoneOutputStream) byteBufferStreamOutput).getOutputStream();
115-
if (outputStream instanceof KeyDataStreamOutput) {
116-
return ((KeyDataStreamOutput) outputStream);
117-
} else if (outputStream instanceof CryptoOutputStream) {
118-
OutputStream wrappedStream =
119-
((CryptoOutputStream) outputStream).getWrappedStream();
120-
if (wrappedStream instanceof KeyDataStreamOutput) {
121-
return ((KeyDataStreamOutput) wrappedStream);
122-
}
123-
} else if (outputStream instanceof CipherOutputStreamOzone) {
124-
OutputStream wrappedStream =
125-
((CipherOutputStreamOzone) outputStream).getWrappedStream();
126-
if (wrappedStream instanceof KeyDataStreamOutput) {
127-
return ((KeyDataStreamOutput) wrappedStream);
128-
}
120+
OutputStream unwrappedStream = unwrap(outputStream);
121+
if (unwrappedStream instanceof KeyDataStreamOutput) {
122+
return ((KeyDataStreamOutput) unwrappedStream);
123+
}
124+
}
125+
// Otherwise return null.
126+
return null;
127+
}
128+
129+
private KeyCommitOutput getKeyCommitOutput() {
130+
if (byteBufferStreamOutput instanceof KeyCommitOutput) {
131+
return (KeyCommitOutput) byteBufferStreamOutput;
132+
}
133+
if (byteBufferStreamOutput instanceof OzoneOutputStream) {
134+
OutputStream outputStream =
135+
((OzoneOutputStream) byteBufferStreamOutput).getOutputStream();
136+
OutputStream unwrappedStream = unwrap(outputStream);
137+
if (unwrappedStream instanceof KeyCommitOutput) {
138+
return (KeyCommitOutput) unwrappedStream;
129139
}
130-
} else if (byteBufferStreamOutput instanceof KeyDataStreamOutput) {
131-
return ((KeyDataStreamOutput) byteBufferStreamOutput);
132140
}
133141
// Otherwise return null.
134142
return null;
135143
}
136144

145+
public void setPreCommits(
146+
List<CheckedRunnable<IOException>> preCommits) {
147+
KeyCommitOutput keyCommitOutput = getKeyCommitOutput();
148+
if (keyCommitOutput != null) {
149+
keyCommitOutput.setPreCommits(preCommits);
150+
return;
151+
}
152+
throw new IllegalStateException(
153+
"Output stream is not backed by KeyCommitOutput: " +
154+
byteBufferStreamOutput.getClass());
155+
}
156+
157+
private static OutputStream unwrap(OutputStream outputStream) {
158+
if (outputStream instanceof CryptoOutputStream) {
159+
return ((CryptoOutputStream) outputStream).getWrappedStream();
160+
} else if (outputStream instanceof CipherOutputStreamOzone) {
161+
return ((CipherOutputStreamOzone) outputStream).getWrappedStream();
162+
}
163+
return outputStream;
164+
}
165+
137166
@Override
138167
public void hflush() throws IOException {
139168
hsync();

‎hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/ObjectEndpointStreaming.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import java.security.DigestInputStream;
3030
import java.security.MessageDigest;
3131
import java.util.ArrayList;
32-
import java.util.Collections;
3332
import java.util.List;
3433
import java.util.Map;
3534
import javax.ws.rs.core.HttpHeaders;
@@ -154,7 +153,7 @@ public static Pair<String, Long> putKeyWithStream(
154153
preCommits.add(checkSha256Hook);
155154
}
156155

157-
streamOutput.getKeyDataStreamOutput().setPreCommits(preCommits);
156+
streamOutput.setPreCommits(preCommits);
158157
}
159158
return Pair.of(md5Hash, writeLen);
160159
}
@@ -237,13 +236,15 @@ public static Response createMultipartKey(OzoneBucket ozoneBucket, String key,
237236
writeToStreamOutput(streamOutput, body, chunkSize, length);
238237
eTag = DatatypeConverter.printHexBinary(
239238
body.getMessageDigest(OzoneConsts.MD5_HASH).digest()).toLowerCase();
239+
List<CheckedRunnable<IOException>> preCommits = new ArrayList<>();
240240
String clientContentMD5 = headers.getHeaderString(S3Consts.CHECKSUM_HEADER);
241241
if (clientContentMD5 != null) {
242242
CheckedRunnable<IOException> checkContentMD5Hook = () -> {
243243
S3Utils.validateContentMD5(clientContentMD5, eTag, key);
244244
};
245-
streamOutput.getKeyDataStreamOutput().setPreCommits(Collections.singletonList(checkContentMD5Hook));
245+
preCommits.add(checkContentMD5Hook);
246246
}
247+
streamOutput.setPreCommits(preCommits);
247248
((KeyMetadataAware)streamOutput).getMetadata().put(OzoneConsts.ETAG, eTag);
248249
METRICS.incPutKeySuccessLength(putLength);
249250
perf.appendMetaLatencyNanos(metadataLatencyNs);

‎hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/OzoneBucketStub.java‎

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -326,6 +326,12 @@ public OzoneDataStreamOutput createMultipartStreamKey(String key,
326326
if (multipartInfo == null || !multipartInfo.getUploadId().equals(uploadID)) {
327327
throw new OMException(ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR);
328328
} else {
329+
if (isECMultipartUpload(multipartInfo)) {
330+
OzoneOutputStream outputStream =
331+
createMultipartKey(key, size, partNumber, uploadID);
332+
return new OzoneDataStreamOutputStub(outputStream, key + size);
333+
}
334+
329335
ByteBufferStreamOutput byteBufferStreamOutput =
330336
new KeyMetadataAwareByteBufferStreamOutput(new HashMap<>()) {
331337
private final ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
@@ -364,6 +370,12 @@ public void write(ByteBuffer b, int off, int len)
364370
}
365371
}
366372

373+
private boolean isECMultipartUpload(MultipartInfoStub multipartInfo) {
374+
ReplicationConfig config = multipartInfo.getReplicationConfig();
375+
return config != null &&
376+
config.getReplicationType() == HddsProtos.ReplicationType.EC;
377+
}
378+
367379
@Override
368380
public OzoneInputStream readKey(String key) throws IOException {
369381
return new OzoneInputStream(new ByteArrayInputStream(keyContents.get(key)));
@@ -502,7 +514,8 @@ public OmMultipartInfo initiateMultipartUpload(String keyName,
502514
ReplicationConfig config, Map<String, String> metadata, Map<String, String> tags)
503515
throws IOException {
504516
String uploadID = UUID.randomUUID().toString();
505-
keyToMultipartUpload.put(keyName, new MultipartInfoStub(uploadID, metadata, tags));
517+
keyToMultipartUpload.put(keyName,
518+
new MultipartInfoStub(uploadID, config, metadata, tags));
506519
return new OmMultipartInfo(getVolumeName(), getName(), keyName, uploadID);
507520
}
508521

@@ -911,12 +924,14 @@ public Map<String, String> getMetadata() {
911924
private static class MultipartInfoStub {
912925

913926
private final String uploadId;
927+
private final ReplicationConfig replicationConfig;
914928
private final Map<String, String> metadata;
915929
private final Map<String, String> tags;
916930

917-
MultipartInfoStub(String uploadId, Map<String, String> metadata,
918-
Map<String, String> tags) {
931+
MultipartInfoStub(String uploadId, ReplicationConfig replicationConfig,
932+
Map<String, String> metadata, Map<String, String> tags) {
919933
this.uploadId = uploadId;
934+
this.replicationConfig = replicationConfig;
920935
this.metadata = metadata;
921936
this.tags = tags;
922937
}
@@ -925,6 +940,10 @@ public String getUploadId() {
925940
return uploadId;
926941
}
927942

943+
public ReplicationConfig getReplicationConfig() {
944+
return replicationConfig;
945+
}
946+
928947
public Map<String, String> getMetadata() {
929948
return metadata;
930949
}

‎hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestPartUpload.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import org.apache.hadoop.ozone.client.OzoneMultipartUploadPartListParts;
5959
import org.apache.hadoop.ozone.s3.exception.OS3Exception;
6060
import org.apache.hadoop.ozone.s3.exception.S3ErrorTable;
61+
import org.apache.hadoop.ozone.s3.util.S3StorageType;
6162
import org.junit.jupiter.api.BeforeEach;
6263
import org.junit.jupiter.api.Test;
6364
import org.junit.jupiter.params.Parameter;
@@ -126,6 +127,42 @@ public void testPartUpload() throws Exception {
126127
}
127128
}
128129

130+
@Test
131+
public void testPartUploadWithStandardIA() throws Exception {
132+
when(headers.getHeaderString(STORAGE_CLASS_HEADER))
133+
.thenReturn(S3StorageType.STANDARD_IA.name(), (String)null);
134+
String keyName = UUID.randomUUID().toString();
135+
String uploadID = initiateMultipartUpload(rest, OzoneConsts.S3_BUCKET, keyName);
136+
137+
String content = "Multipart Upload";
138+
try (Response response = put(rest, OzoneConsts.S3_BUCKET, keyName, 1, uploadID, content)) {
139+
assertNotNull(response.getHeaderString(OzoneConsts.ETAG));
140+
assertEquals(200, response.getStatus());
141+
}
142+
assertContentLength(uploadID, keyName, content.length());
143+
}
144+
145+
@Test
146+
public void testPartUploadWithStandardIAAndContentMD5() throws Exception {
147+
when(headers.getHeaderString(STORAGE_CLASS_HEADER))
148+
.thenReturn(S3StorageType.STANDARD_IA.name(), (String)null);
149+
String content = "Multipart Upload Part";
150+
byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
151+
byte[] md5Bytes = MessageDigest.getInstance("MD5").digest(contentBytes);
152+
String md5Base64 = Base64.getEncoder().encodeToString(md5Bytes);
153+
when(headers.getHeaderString("Content-MD5")).thenReturn(md5Base64);
154+
155+
String keyName = UUID.randomUUID().toString();
156+
String uploadID = initiateMultipartUpload(rest, OzoneConsts.S3_BUCKET, keyName);
157+
158+
try (Response response = put(rest, OzoneConsts.S3_BUCKET, keyName, 1,
159+
uploadID, content)) {
160+
assertNotNull(response.getHeaderString(OzoneConsts.ETAG));
161+
assertEquals(200, response.getStatus());
162+
}
163+
assertContentLength(uploadID, keyName, content.length());
164+
}
165+
129166
@Test
130167
public void testPartUploadWithIncorrectUploadID() {
131168
assertErrorResponse(S3ErrorTable.NO_SUCH_UPLOAD,

0 commit comments

Comments
 (0)