FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

fix: update ApiaryUnbufferedWritableByteChannel to be graceful of non-quantum aligned write calls by BenWhitehead · Pull Request #2493 · googleapis/java-storage · GitHub

/ java-storage Public

fix: update ApiaryUnbufferedWritableByteChannel to be graceful of non-quantum aligned write calls #2493

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
BenWhitehead merged 1 commit into from
Apr 9, 2024
Conversation 6 Commits 1 Checks 0 Files changed
Merged

fix: update ApiaryUnbufferedWritableByteChannel to be graceful of non-quantum aligned write calls #2493

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension .java  (6) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
16 changes: 9 additions & 7 deletions ...d-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedWritableByteChannel.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -96,30 +96,32 @@ private long internalWrite(ByteBuffer[] srcs, int offset, int length, boolean fi
}
RewindableContent content = RewindableContent.of(Utils.subArray(srcs, offset, length));
long available = content.getLength();
// as long as request has at least 256KiB GCS will accept bytes in 256KiB increments,
// however if a request is smaller than 256KiB it MUST be the finalization request.
if (!finalize && available < ByteSizeConstants._256KiB) {
Comment thread
BenWhitehead marked this conversation as resolved.
return 0;
}
long newFinalByteOffset = cumulativeByteCount + available;
final HttpContentRange header;
ByteRangeSpec rangeSpec = ByteRangeSpec.explicit(cumulativeByteCount, newFinalByteOffset);
boolean quantumAligned = available % ByteSizeConstants._256KiB == 0;
if (quantumAligned && finalize) {
if (finalize) {
header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
finished = true;
} else if (quantumAligned) {
} else {
header = HttpContentRange.of(rangeSpec);
} else { // not quantum aligned, have to finalize
header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
finished = true;
}
try {
ResumableOperationResult<@Nullable StorageObject> operationResult =
session.put(content, header);
long persistedSize = operationResult.getPersistedSize();
committedBytesCallback.accept(persistedSize);
long written = persistedSize - cumulativeByteCount;
this.cumulativeByteCount = persistedSize;
if (finished) {
StorageObject storageObject = operationResult.getObject();
result.set(storageObject);
}
return available;
return written;
} catch (Exception e) {
result.setException(e);
throw StorageException.coalesce(e);
Expand Down
3 changes: 1 addition & 2 deletions ...d-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,7 @@ enum JsonResumableSessionFailureScenario {
SCENARIO_7(
BaseServiceException.UNKNOWN_CODE,
"dataLoss",
"Client side data loss detected. Bytes acked is more than client sent."),
SCENARIO_9(503, "backendNotConnected", "Ack less than bytes sent");
"Client side data loss detected. Bytes acked is more than client sent.");

private static final String PREFIX_I = "\t|< ";
private static final String PREFIX_O = "\t|> ";
Expand Down
7 changes: 3 additions & 4 deletions google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -109,10 +109,9 @@ public void rewindTo(long offset) {
success = true;
return ResumableOperationResult.incremental(ackRange.endOffset());
} else if (ackRange.endOffset() < effectiveEnd) {
StorageException se =
JsonResumableSessionFailureScenario.SCENARIO_9.toStorageException(uploadId, response);
span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
throw se;
rewindTo(ackRange.endOffset());
success = true;
return ResumableOperationResult.incremental(ackRange.endOffset());
} else {
StorageException se =
JsonResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response);
Expand Down
45 changes: 0 additions & 45 deletions ...oud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -110,51 +110,6 @@ public void emptyObjectHappyPath() throws Exception {
}
}

/**
*
*
* <h4>S.9</h4>
*
* Partial successful append to session
*
* <p>The client has sent N bytes, the server confirmed N bytes as committed. The client sends K
* bytes starting at offset N. The server responds with only N + L with 0 &lt;= L &lt; K bytes as
* committed.
*/
@Test
public void scenario9() throws Exception {

BenWhitehead Apr 8, 2024

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

This test would verify that GCS ack'ing fewer bytes than were sent was correctly categorized as a failure scenario, however as we are now gracefully handling the partial write case. We no longer need this test for failure categorization.

HttpRequestHandler handler =
req -> {
String contentRangeString = req.headers().get(CONTENT_RANGE);
HttpContentRange parse = HttpContentRange.parse(contentRangeString);
long endInclusive = ((HttpContentRange.HasRange<?>) parse).range().endOffsetInclusive();
FullHttpResponse resp =
new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive - 1);
resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
return resp;
};

try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
URI endpoint = fakeHttpServer.getEndpoint();
String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());

AtomicLong confirmedBytes = new AtomicLong(-1L);

JsonResumableSessionPutTask task =
new JsonResumableSessionPutTask(
httpClientContext,
uploadUrl,
RewindableContent.empty(),
HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L)));

StorageException se = assertThrows(StorageException.class, task::call);
assertThat(se.getCode()).isEqualTo(503);
assertThat(confirmedBytes.get()).isEqualTo(-1L);
}
}

/**
*
*
Expand Down
2 changes: 1 addition & 1 deletion google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ public void rewindWillQueryStatusOnlyWhenDirty() throws Exception {
DefaultFullHttpResponse resp =
new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
if (range1.getHeaderValue().equals(contentRange)) {
resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
return new DefaultFullHttpResponse(req.protocolVersion(), SERVICE_UNAVAILABLE);
} else if (range2.getHeaderValue().equals(contentRange)) {
resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
} else {
Expand Down
117 changes: 117 additions & 0 deletions ...cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
@@ -0,0 +1,117 @@
/*
* Copyright 2024 Google LLC
*
* Licensed 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 com.google.cloud.storage;

import static com.google.common.truth.Truth.assertThat;

import com.google.api.core.ApiFutures;
import com.google.api.services.storage.model.StorageObject;
import com.google.cloud.storage.ITUnbufferedResumableUploadTest.ObjectSizes;
import com.google.cloud.storage.TransportCompatibility.Transport;
import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
import com.google.cloud.storage.UnifiedOpts.Opts;
import com.google.cloud.storage.it.runner.StorageITRunner;
import com.google.cloud.storage.it.runner.annotations.Backend;
import com.google.cloud.storage.it.runner.annotations.CrossRun;
import com.google.cloud.storage.it.runner.annotations.CrossRun.Exclude;
import com.google.cloud.storage.it.runner.annotations.Inject;
import com.google.cloud.storage.it.runner.annotations.Parameterized;
import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider;
import com.google.cloud.storage.it.runner.registry.Generator;
import com.google.cloud.storage.spi.v1.StorageRpc;
import com.google.common.collect.ImmutableList;
import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import org.junit.Test;
import org.junit.runner.RunWith;

@RunWith(StorageITRunner.class)
@CrossRun(
backends = {Backend.PROD},
transports = {Transport.HTTP, Transport.GRPC})
@Parameterized(ObjectSizes.class)
public final class ITUnbufferedResumableUploadTest {

@Inject public Storage storage;
@Inject public BucketInfo bucket;
@Inject public Generator generator;

@Parameter public int objectSize;

public static final class ObjectSizes implements ParametersProvider {

@Override
public ImmutableList<Integer> parameters() {
return ImmutableList.of(256 * 1024, 2 * 1024 * 1024);
}
}

@Test
@Exclude(transports = Transport.GRPC)
public void json()
throws IOException, ExecutionException, InterruptedException, TimeoutException {
BlobInfo blobInfo = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
Opts<ObjectTargetOpt> opts = Opts.empty();
final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
BlobInfo updated = opts.blobInfoMapper().apply(builder).build();

StorageObject encode = Conversions.json().blobInfo().encode(updated);
HttpStorageOptions options = (HttpStorageOptions) storage.getOptions();
Supplier<String> uploadIdSupplier =
ResumableMedia.startUploadForBlobInfo(
options,
updated,
optionsMap,
StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler());
JsonResumableWrite jsonResumableWrite =
JsonResumableWrite.of(encode, optionsMap, uploadIdSupplier.get(), 0);

UnbufferedWritableByteChannelSession<StorageObject> session =
ResumableMedia.http()
.write()
.byteChannel(HttpClientContext.from(options.getStorageRpcV1()))
.resumable()
.unbuffered()
.setStartAsync(ApiFutures.immediateFuture(jsonResumableWrite))
.build();

int additional = 13;
long size = objectSize + additional;
ByteBuffer b = DataGenerator.base64Characters().genByteBuffer(size);

UnbufferedWritableByteChannel open = session.open();
int written = open.write(b);
assertThat(written).isEqualTo(objectSize);
assertThat(b.remaining()).isEqualTo(additional);

int writtenAndClose = open.writeAndClose(b);
assertThat(writtenAndClose).isEqualTo(additional);
Comment thread
BenWhitehead marked this conversation as resolved.
open.close();

StorageObject storageObject = session.getResult().get(2, TimeUnit.SECONDS);
assertThat(storageObject.getSize()).isEqualTo(BigInteger.valueOf(size));
}
}
Back | FazBrowse Home | New Git URL