Skip to content

Commit ddf9add

Browse files
Dhriti07Dhriti Chopra
Dhriti07
and
Dhriti Chopra
authored
feat(storage): allow checksum on appendable upload finalization (#13833)
Allow checksum on appendable upload finalization --------- Co-authored-by: Dhriti Chopra <dhritichopra@google.com>
1 parent 1b7c2e0 commit ddf9add

6 files changed

Lines changed: 330 additions & 8 deletions

File tree

‎java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import java.util.concurrent.ExecutionException;
2727
import java.util.concurrent.TimeUnit;
2828
import java.util.concurrent.TimeoutException;
29+
import org.checkerframework.checker.nullness.qual.Nullable;
2930

3031
final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
3132

@@ -36,6 +37,7 @@ final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWri
3637
private boolean open;
3738
private long writeOffset;
3839
private volatile boolean nextWriteShouldFinalize;
40+
private @Nullable String expectedCrc32c;
3941
private boolean writeCalledAtLeastOnce;
4042
private long lastFlushOffset;
4143

@@ -53,6 +55,7 @@ final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWri
5355
this.open = true;
5456
this.writeOffset = writeOffset;
5557
this.nextWriteShouldFinalize = false;
58+
this.expectedCrc32c = null;
5659
this.writeThrewError = false;
5760
this.lastFlushOffset = writeOffset;
5861
}
@@ -96,7 +99,7 @@ public void close() throws IOException {
9699
}
97100
if (nextWriteShouldFinalize) {
98101
//noinspection StatementWithEmptyBody
99-
while (!stream.finishWrite(writeOffset)) {}
102+
while (!stream.finishWrite(writeOffset, expectedCrc32c)) {}
100103
} else {
101104
//noinspection StatementWithEmptyBody
102105
while (!stream.closeStream(writeOffset)) {}
@@ -113,6 +116,11 @@ public void nextWriteShouldFinalize() {
113116
this.nextWriteShouldFinalize = true;
114117
}
115118

119+
public void nextWriteShouldFinalize(@Nullable String expectedCrc32c) {
120+
this.nextWriteShouldFinalize = true;
121+
this.expectedCrc32c = expectedCrc32c;
122+
}
123+
116124
void flush() throws InterruptedException {
117125
stream.flush();
118126
stream.awaitAckOf(writeOffset);
@@ -155,7 +163,7 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
155163
if (i < lastIdx && !shouldFlush) {
156164
appended = stream.append(datum);
157165
} else if (i == lastIdx && remainingAfterPacking == 0 && nextWriteShouldFinalize) {
158-
appended = stream.appendAndFinalize(datum);
166+
appended = stream.appendAndFinalize(datum, expectedCrc32c);
159167
} else {
160168
appended = stream.appendAndFlush(datum);
161169
}

‎java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -126,12 +126,13 @@ public boolean appendAndFlush(ChunkSegmenter.@NonNull ChunkSegment data) {
126126
}
127127
}
128128

129-
public boolean appendAndFinalize(ChunkSegmenter.@NonNull ChunkSegment data) {
129+
public boolean appendAndFinalize(
130+
ChunkSegmenter.@NonNull ChunkSegment data, @Nullable String expectedCrc32c) {
130131
lock.lock();
131132
try {
132133
boolean offered = state.offer(data);
133134
if (offered) {
134-
finishWrite(state.getTotalSentBytes());
135+
finishWrite(state.getTotalSentBytes(), expectedCrc32c);
135136
}
136137
return offered;
137138
} finally {
@@ -163,6 +164,10 @@ public void flush() {
163164
}
164165

165166
public boolean finishWrite(long length) {
167+
return finishWrite(length, null);
168+
}
169+
170+
public boolean finishWrite(long length, @Nullable String expectedCrc32c) {
166171
lock.lock();
167172
try {
168173
// if we're already finalizing, ack rather than enqueueing again
@@ -172,10 +177,15 @@ public boolean finishWrite(long length) {
172177

173178
BidiWriteObjectRequest.Builder b =
174179
BidiWriteObjectRequest.newBuilder().setWriteOffset(length).setFinishWrite(true);
175-
Crc32cLengthKnown cumulativeCrc32c = state.getCumulativeCrc32c();
176-
if (cumulativeCrc32c != null) {
177-
b.setObjectChecksums(
178-
ObjectChecksums.newBuilder().setCrc32C(cumulativeCrc32c.getValue()).build());
180+
if (expectedCrc32c != null) {
181+
int crc32cInt = Utils.crc32cCodec.decode(expectedCrc32c);
182+
b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cInt).build());
183+
} else {
184+
Crc32cLengthKnown cumulativeCrc32c = state.getCumulativeCrc32c();
185+
if (cumulativeCrc32c != null) {
186+
b.setObjectChecksums(
187+
ObjectChecksums.newBuilder().setCrc32C(cumulativeCrc32c.getValue()).build());
188+
}
179189
}
180190
BidiWriteObjectRequest msg = b.build();
181191
boolean offer = state.offer(msg);

‎java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobAppendableUpload.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import java.nio.channels.ClosedChannelException;
2727
import java.nio.channels.WritableByteChannel;
2828
import java.util.concurrent.TimeUnit;
29+
import org.checkerframework.checker.nullness.qual.Nullable;
2930

3031
/**
3132
* Interface representing those methods which can be used to write to and interact with an
@@ -158,6 +159,32 @@ interface AppendableUploadWriteableByteChannel extends WritableByteChannel {
158159
@BetaApi
159160
void finalizeAndClose() throws IOException;
160161

162+
/**
163+
* <b>This method is blocking</b>
164+
*
165+
* <p>Finalize the upload and close this instance to further {@link #write(ByteBuffer)}ing. This
166+
* will close any underlying stream and release any releasable resources once out of scope.
167+
*
168+
* <p>Once this method is called, and returns no more writes to the object will be allowed by
169+
* GCS.
170+
*
171+
* <p>This method and {@link #close()} are mutually exclusive. If one of the other methods are
172+
* called before this method, this method will be a no-op.
173+
*
174+
* @param expectedCrc32c A Base64 encoded string representing the expected CRC32c value for the
175+
* entire object. If provided, the server will validate the final object's CRC32c against
176+
* this value. If there's a mismatch, the server will return an error (such as
177+
* InvalidArgument), and this method will throw a {@link StorageException} or equivalent
178+
* exception, failing the upload.
179+
* @see Storage#blobAppendableUpload(BlobInfo, BlobAppendableUploadConfig, BlobWriteOption...)
180+
* @see BlobAppendableUploadConfig.CloseAction#FINALIZE_WHEN_CLOSING
181+
* @see BlobAppendableUploadConfig#getCloseAction()
182+
* @see BlobAppendableUploadConfig#withCloseAction(CloseAction)
183+
* @since 2.51.0 This new api is in preview and is subject to breaking changes.
184+
*/
185+
@BetaApi
186+
void finalizeAndClose(@Nullable String expectedCrc32c) throws IOException;
187+
161188
/**
162189
* <b>This method is blocking</b>
163190
*
@@ -197,5 +224,28 @@ interface AppendableUploadWriteableByteChannel extends WritableByteChannel {
197224
*/
198225
@BetaApi
199226
void close() throws IOException;
227+
228+
/**
229+
* <b>This method is blocking</b>
230+
*
231+
* <p>Close this instance to further {@link #write(ByteBuffer)}ing.
232+
*
233+
* <p>This method behaves like {@link #close()}, but allows passing an optional expected CRC32c
234+
* checksum. If expectedCrc32c is null, it behaves identically to {@link #close()}.
235+
*
236+
* @param expectedCrc32c A Base64 encoded string representing the expected CRC32c value for the
237+
* entire object. If provided, the server will validate the final object's CRC32c against
238+
* this value. If there's a mismatch, the server will return an error (such as
239+
* InvalidArgument), and this method will throw a {@link StorageException} or equivalent
240+
* exception, failing the upload. May be null.
241+
* @throws IllegalArgumentException if the stream was not configured with {@link
242+
* CloseAction#FINALIZE_WHEN_CLOSING} and expectedCrc32c is not null.
243+
* @see Storage#blobAppendableUpload(BlobInfo, BlobAppendableUploadConfig, BlobWriteOption...)
244+
* @see BlobAppendableUploadConfig#getCloseAction()
245+
* @see BlobAppendableUploadConfig#withCloseAction(CloseAction)
246+
* @since 2.51.0 This new api is in preview and is subject to breaking changes.
247+
*/
248+
@BetaApi
249+
void close(@Nullable String expectedCrc32c) throws IOException;
200250
}
201251
}

‎java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobAppendableUploadImpl.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import java.io.InterruptedIOException;
2525
import java.nio.ByteBuffer;
2626
import java.util.concurrent.locks.ReentrantLock;
27+
import org.checkerframework.checker.nullness.qual.Nullable;
2728

2829
@BetaApi
2930
final class BlobAppendableUploadImpl implements BlobAppendableUpload {
@@ -132,6 +133,19 @@ public void finalizeAndClose() throws IOException {
132133
}
133134
}
134135

136+
@Override
137+
public void finalizeAndClose(@Nullable String expectedCrc32c) throws IOException {
138+
lock.lock();
139+
try {
140+
if (buffered.isOpen()) {
141+
unbuffered.nextWriteShouldFinalize(expectedCrc32c);
142+
buffered.close();
143+
}
144+
} finally {
145+
lock.unlock();
146+
}
147+
}
148+
135149
@Override
136150
public void closeWithoutFinalizing() throws IOException {
137151
lock.lock();
@@ -152,5 +166,17 @@ public void close() throws IOException {
152166
closeWithoutFinalizing();
153167
}
154168
}
169+
170+
@Override
171+
public void close(@Nullable String expectedCrc32c) throws IOException {
172+
if (finalizeOnClose) {
173+
finalizeAndClose(expectedCrc32c);
174+
} else if (expectedCrc32c == null) {
175+
closeWithoutFinalizing();
176+
} else {
177+
throw new IllegalArgumentException(
178+
"expectedCrc32c can only be provided when finalizeOnClose is true.");
179+
}
180+
}
155181
}
156182
}

‎java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2216,6 +2216,32 @@ public void finalizeAndClose() throws IOException {
22162216
}
22172217
}
22182218

2219+
@Override
2220+
@BetaApi
2221+
public void finalizeAndClose(@Nullable String expectedCrc32c) throws IOException {
2222+
try (Scope ignore = openSpan.makeCurrent()) {
2223+
Span span = tracer.spanBuilder("finalizeAndClose").startSpan();
2224+
try (Scope ignore2 = span.makeCurrent()) {
2225+
delegate.finalizeAndClose(expectedCrc32c);
2226+
} catch (Throwable t) {
2227+
span.recordException(t);
2228+
span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
2229+
throw t;
2230+
} finally {
2231+
span.end();
2232+
}
2233+
} catch (IOException | RuntimeException e) {
2234+
openSpan.recordException(e);
2235+
openSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2236+
uploadSpan.recordException(e);
2237+
uploadSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2238+
throw e;
2239+
} finally {
2240+
openSpan.end();
2241+
uploadSpan.end();
2242+
}
2243+
}
2244+
22192245
@Override
22202246
@BetaApi
22212247
public void closeWithoutFinalizing() throws IOException {
@@ -2268,6 +2294,32 @@ public void close() throws IOException {
22682294
}
22692295
}
22702296

2297+
@Override
2298+
@BetaApi
2299+
public void close(@Nullable String expectedCrc32c) throws IOException {
2300+
try (Scope ignore = openSpan.makeCurrent()) {
2301+
Span span = tracer.spanBuilder("close").startSpan();
2302+
try (Scope ignore2 = span.makeCurrent()) {
2303+
delegate.close(expectedCrc32c);
2304+
} catch (Throwable t) {
2305+
span.recordException(t);
2306+
span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
2307+
throw t;
2308+
} finally {
2309+
span.end();
2310+
}
2311+
} catch (IOException | RuntimeException e) {
2312+
openSpan.recordException(e);
2313+
openSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2314+
uploadSpan.recordException(e);
2315+
uploadSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2316+
throw e;
2317+
} finally {
2318+
openSpan.end();
2319+
uploadSpan.end();
2320+
}
2321+
}
2322+
22712323
@Override
22722324
public void flush() throws IOException {
22732325
try (Scope ignore = openSpan.makeCurrent()) {

0 commit comments

Comments
 (0)