Skip to content

Commit 470b8cd

Browse files
BenWhiteheadBenWhitehead
authored
feat: update all automatic retry behavior to be idempotency aware (#1132)
1. Add new configuration option `StorageOptions.Builder#setStorageRetryStrategy` 2. Add new interface `StorageRetryStrategy` 3. Update all calls to be made idempotency aware 4. Change default behavior of automatic retries to only retry those operations which are deemed to be idempotent 5. Add @deprecated method `StorageRetryStrategy.getLegacyStorageRetryStrategy` providing access to a strategy which matches the previous sometimes unsafe behavior 6. make BlobWriteChannelTest cases idempotent 7. make StorageImplMockitoTest#testCreateBlobRetry idempotent
1 parent 2d64f90 commit 470b8cd

File tree

6 files changed

+58
-38
lines changed

6 files changed

+58
-38
lines changed

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ public Builder setTransportOptions(TransportOptions transportOptions) {
8787
* @return the builder
8888
* @see StorageRetryStrategy#getDefaultStorageRetryStrategy()
8989
*/
90-
Builder setStorageRetryStrategy(StorageRetryStrategy storageRetryStrategy) {
90+
public Builder setStorageRetryStrategy(StorageRetryStrategy storageRetryStrategy) {
9191
this.storageRetryStrategy =
9292
requireNonNull(storageRetryStrategy, "storageRetryStrategy must be non null");
9393
return this;
@@ -125,7 +125,7 @@ public TransportOptions getDefaultTransportOptions() {
125125
}
126126

127127
public StorageRetryStrategy getStorageRetryStrategy() {
128-
return StorageRetryStrategy.getLegacyStorageRetryStrategy();
128+
return StorageRetryStrategy.getDefaultStorageRetryStrategy();
129129
}
130130
}
131131

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
* @see #getDefaultStorageRetryStrategy()
3030
* @see #getUniformStorageRetryStrategy()
3131
*/
32-
interface StorageRetryStrategy extends Serializable {
32+
public interface StorageRetryStrategy extends Serializable {
3333

3434
/**
3535
* Factory method to provide a {@link ResultRetryAlgorithm} which will be used to evaluate whether

‎google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java‎

Lines changed: 47 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -65,8 +65,12 @@ public class BlobWriteChannelTest {
6565
private static final String BLOB_NAME = "n";
6666
private static final String UPLOAD_ID = "uploadid";
6767
private static final BlobInfo BLOB_INFO = BlobInfo.newBuilder(BUCKET_NAME, BLOB_NAME).build();
68+
private static final BlobInfo BLOB_INFO_WITH_GENERATION =
69+
BlobInfo.newBuilder(BUCKET_NAME, BLOB_NAME, 1L).build();
6870
private static final StorageObject UPDATED_BLOB = new StorageObject();
6971
private static final Map<StorageRpc.Option, ?> EMPTY_RPC_OPTIONS = ImmutableMap.of();
72+
private static final Map<StorageRpc.Option, ?> RPC_OPTIONS_GENERATION =
73+
ImmutableMap.of(StorageRpc.Option.IF_GENERATION_MATCH, 1L);
7074
private static final int MIN_CHUNK_SIZE = 256 * 1024;
7175
private static final int DEFAULT_CHUNK_SIZE = 60 * MIN_CHUNK_SIZE; // 15MiB
7276
private static final int CUSTOM_CHUNK_SIZE = 4 * MIN_CHUNK_SIZE;
@@ -112,11 +116,12 @@ public void testCreate() {
112116

113117
@Test
114118
public void testCreateRetryableError() {
115-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS))
119+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
116120
.andThrow(socketClosedException);
117-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
121+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
122+
.andReturn(UPLOAD_ID);
118123
replay(storageRpcMock);
119-
writer = newWriter();
124+
writer = newWriter(true);
120125
assertTrue(writer.isOpen());
121126
assertNull(writer.getStorageObject());
122127
}
@@ -146,7 +151,8 @@ public void testWriteWithoutFlush() throws Exception {
146151
public void testWriteWithFlushRetryChunk() throws Exception {
147152
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
148153
Capture<byte[]> capturedBuffer = Capture.newInstance();
149-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
154+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
155+
.andReturn(UPLOAD_ID);
150156
expect(
151157
storageRpcMock.writeWithResponse(
152158
eq(UPLOAD_ID),
@@ -167,7 +173,7 @@ public void testWriteWithFlushRetryChunk() throws Exception {
167173
eq(false)))
168174
.andReturn(null);
169175
replay(storageRpcMock);
170-
writer = newWriter();
176+
writer = newWriter(true);
171177
writer.setChunkSize(MIN_CHUNK_SIZE);
172178
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
173179
assertTrue(writer.isOpen());
@@ -179,7 +185,8 @@ public void testWriteWithFlushRetryChunk() throws Exception {
179185
public void testWriteWithRetryFullChunk() throws Exception {
180186
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
181187
Capture<byte[]> capturedBuffer = Capture.newInstance();
182-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
188+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
189+
.andReturn(UPLOAD_ID);
183190
expect(
184191
storageRpcMock.writeWithResponse(
185192
eq(UPLOAD_ID), (byte[]) anyObject(), eq(0), eq(0L), eq(MIN_CHUNK_SIZE), eq(false)))
@@ -204,7 +211,7 @@ public void testWriteWithRetryFullChunk() throws Exception {
204211
eq(true)))
205212
.andReturn(BLOB_INFO.toPb());
206213
replay(storageRpcMock);
207-
writer = newWriter();
214+
writer = newWriter(true);
208215
writer.setChunkSize(MIN_CHUNK_SIZE);
209216
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
210217
writer.close();
@@ -217,7 +224,8 @@ public void testWriteWithRetryFullChunk() throws Exception {
217224
public void testWriteWithRemoteProgressMade() throws Exception {
218225
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
219226
Capture<byte[]> capturedBuffer = Capture.newInstance();
220-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
227+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
228+
.andReturn(UPLOAD_ID);
221229
expect(
222230
storageRpcMock.writeWithResponse(
223231
eq(UPLOAD_ID),
@@ -239,7 +247,7 @@ public void testWriteWithRemoteProgressMade() throws Exception {
239247
eq(false)))
240248
.andReturn(null);
241249
replay(storageRpcMock);
242-
writer = newWriter();
250+
writer = newWriter(true);
243251
writer.setChunkSize(MIN_CHUNK_SIZE);
244252
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
245253
assertTrue(writer.isOpen());
@@ -251,7 +259,8 @@ public void testWriteWithRemoteProgressMade() throws Exception {
251259
public void testWriteWithDriftRetryCase4() throws Exception {
252260
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
253261
Capture<byte[]> capturedBuffer = Capture.newInstance();
254-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
262+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
263+
.andReturn(UPLOAD_ID);
255264
expect(
256265
storageRpcMock.writeWithResponse(
257266
eq(UPLOAD_ID),
@@ -272,7 +281,7 @@ public void testWriteWithDriftRetryCase4() throws Exception {
272281
eq(false)))
273282
.andReturn(null);
274283
replay(storageRpcMock);
275-
writer = newWriter();
284+
writer = newWriter(true);
276285
writer.setChunkSize(MIN_CHUNK_SIZE);
277286
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
278287
assertArrayEquals(buffer.array(), capturedBuffer.getValue());
@@ -288,7 +297,8 @@ public void testWriteWithDriftRetryCase4() throws Exception {
288297
public void testWriteWithUnreachableRemoteOffset() throws Exception {
289298
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
290299
Capture<byte[]> capturedBuffer = Capture.newInstance();
291-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
300+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
301+
.andReturn(UPLOAD_ID);
292302
expect(
293303
storageRpcMock.writeWithResponse(
294304
eq(UPLOAD_ID),
@@ -300,7 +310,7 @@ public void testWriteWithUnreachableRemoteOffset() throws Exception {
300310
.andThrow(socketClosedException);
301311
expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(MIN_CHUNK_SIZE + 10L);
302312
replay(storageRpcMock);
303-
writer = newWriter();
313+
writer = newWriter(true);
304314
writer.setChunkSize(MIN_CHUNK_SIZE);
305315
try {
306316
writer.write(buffer);
@@ -317,7 +327,8 @@ public void testWriteWithUnreachableRemoteOffset() throws Exception {
317327
public void testWriteWithRetryAndObjectMetadata() throws Exception {
318328
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
319329
Capture<byte[]> capturedBuffer = Capture.newInstance();
320-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
330+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
331+
.andReturn(UPLOAD_ID);
321332
expect(
322333
storageRpcMock.writeWithResponse(
323334
eq(UPLOAD_ID),
@@ -345,7 +356,7 @@ public void testWriteWithRetryAndObjectMetadata() throws Exception {
345356
expect(storageRpcMock.queryCompletedResumableUpload(eq(UPLOAD_ID), eq((long) MIN_CHUNK_SIZE)))
346357
.andReturn(BLOB_INFO.toPb().setSize(BigInteger.valueOf(MIN_CHUNK_SIZE)));
347358
replay(storageRpcMock);
348-
writer = newWriter();
359+
writer = newWriter(true);
349360
writer.setChunkSize(MIN_CHUNK_SIZE);
350361
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
351362
writer.close();
@@ -358,7 +369,8 @@ public void testWriteWithRetryAndObjectMetadata() throws Exception {
358369
public void testWriteWithUploadCompletedByAnotherClient() throws Exception {
359370
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
360371
Capture<byte[]> capturedBuffer = Capture.newInstance();
361-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
372+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
373+
.andReturn(UPLOAD_ID);
362374
expect(
363375
storageRpcMock.writeWithResponse(
364376
eq(UPLOAD_ID),
@@ -380,7 +392,7 @@ public void testWriteWithUploadCompletedByAnotherClient() throws Exception {
380392
expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
381393
expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
382394
replay(storageRpcMock);
383-
writer = newWriter();
395+
writer = newWriter(true);
384396
writer.setChunkSize(MIN_CHUNK_SIZE);
385397
try {
386398
writer.write(buffer);
@@ -399,7 +411,8 @@ public void testWriteWithUploadCompletedByAnotherClient() throws Exception {
399411
public void testWriteWithLocalOffsetGoingBeyondRemoteOffset() throws Exception {
400412
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
401413
Capture<byte[]> capturedBuffer = Capture.newInstance();
402-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
414+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
415+
.andReturn(UPLOAD_ID);
403416
expect(
404417
storageRpcMock.writeWithResponse(
405418
eq(UPLOAD_ID),
@@ -420,7 +433,7 @@ public void testWriteWithLocalOffsetGoingBeyondRemoteOffset() throws Exception {
420433
.andThrow(socketClosedException);
421434
expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(0L);
422435
replay(storageRpcMock);
423-
writer = newWriter();
436+
writer = newWriter(true);
424437
writer.setChunkSize(MIN_CHUNK_SIZE);
425438
try {
426439
writer.write(buffer);
@@ -437,7 +450,8 @@ public void testWriteWithLocalOffsetGoingBeyondRemoteOffset() throws Exception {
437450
public void testGetCurrentUploadOffset() throws Exception {
438451
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
439452
Capture<byte[]> capturedBuffer = Capture.newInstance();
440-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
453+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
454+
.andReturn(UPLOAD_ID);
441455
expect(
442456
storageRpcMock.writeWithResponse(
443457
eq(UPLOAD_ID),
@@ -468,7 +482,7 @@ public void testGetCurrentUploadOffset() throws Exception {
468482
eq(true)))
469483
.andReturn(BLOB_INFO.toPb());
470484
replay(storageRpcMock);
471-
writer = newWriter();
485+
writer = newWriter(true);
472486
writer.setChunkSize(MIN_CHUNK_SIZE);
473487
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
474488
writer.close();
@@ -481,7 +495,8 @@ public void testGetCurrentUploadOffset() throws Exception {
481495
public void testWriteWithLastFlushRetryChunkButCompleted() throws Exception {
482496
ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
483497
Capture<byte[]> capturedBuffer = Capture.newInstance();
484-
expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
498+
expect(storageRpcMock.open(BLOB_INFO_WITH_GENERATION.toPb(), RPC_OPTIONS_GENERATION))
499+
.andReturn(UPLOAD_ID);
485500
expect(
486501
storageRpcMock.writeWithResponse(
487502
eq(UPLOAD_ID),
@@ -495,7 +510,7 @@ public void testWriteWithLastFlushRetryChunkButCompleted() throws Exception {
495510
expect(storageRpcMock.queryCompletedResumableUpload(eq(UPLOAD_ID), eq((long) MIN_CHUNK_SIZE)))
496511
.andReturn(BLOB_INFO.toPb().setSize(BigInteger.valueOf(MIN_CHUNK_SIZE)));
497512
replay(storageRpcMock);
498-
writer = newWriter();
513+
writer = newWriter(true);
499514
assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
500515
writer.close();
501516
assertFalse(writer.isRetrying());
@@ -825,17 +840,23 @@ public void testSaveAndRestoreWithSignedURL() throws Exception {
825840
}
826841

827842
private BlobWriteChannel newWriter() {
828-
Map<StorageRpc.Option, ?> optionsMap = EMPTY_RPC_OPTIONS;
843+
return newWriter(false);
844+
}
845+
846+
private BlobWriteChannel newWriter(boolean withGeneration) {
847+
Map<StorageRpc.Option, ?> optionsMap =
848+
withGeneration ? RPC_OPTIONS_GENERATION : EMPTY_RPC_OPTIONS;
829849
ResultRetryAlgorithm<?> createResultAlgorithm =
830850
retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap);
831851
ResultRetryAlgorithm<?> writeResultAlgorithm =
832852
retryAlgorithmManager.getForResumableUploadSessionWrite(optionsMap);
853+
final BlobInfo blobInfo = withGeneration ? BLOB_INFO_WITH_GENERATION : BLOB_INFO;
833854
return BlobWriteChannel.newBuilder()
834855
.setStorageOptions(options)
835-
.setBlobInfo(BLOB_INFO)
856+
.setBlobInfo(blobInfo)
836857
.setUploadIdSupplier(
837858
ResumableMedia.startUploadForBlobInfo(
838-
options, BLOB_INFO, optionsMap, createResultAlgorithm))
859+
options, blobInfo, optionsMap, createResultAlgorithm))
839860
.setAlgorithmForWrite(writeResultAlgorithm)
840861
.build();
841862
}

‎google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -38,9 +38,4 @@ public static Blob blobCopyWithStorage(Blob b, Storage s) {
3838
BlobInfo.BuilderImpl builder = (BlobInfo.BuilderImpl) BlobInfo.fromPb(b.toPb()).toBuilder();
3939
return new Blob(s, builder);
4040
}
41-
42-
public static StorageOptions.Builder useDefaultStorageRetryStrategy(
43-
StorageOptions.Builder builder) {
44-
return builder.setStorageRetryStrategy(StorageRetryStrategy.getDefaultStorageRetryStrategy());
45-
}
4641
}

‎google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import com.google.cloud.ServiceOptions;
3737
import com.google.cloud.Tuple;
3838
import com.google.cloud.WriteChannel;
39+
import com.google.cloud.storage.Storage.BlobTargetOption;
3940
import com.google.cloud.storage.spi.StorageRpcFactory;
4041
import com.google.cloud.storage.spi.v1.StorageRpc;
4142
import com.google.common.collect.ImmutableList;
@@ -121,6 +122,8 @@ public class StorageImplMockitoTest {
121122

122123
// Empty StorageRpc options
123124
private static final Map<StorageRpc.Option, ?> EMPTY_RPC_OPTIONS = ImmutableMap.of();
125+
private static final Map<StorageRpc.Option, ?> BLOB_INFO1_RPC_OPTIONS_WITH_GENERATION =
126+
ImmutableMap.of(StorageRpc.Option.IF_GENERATION_MATCH, 24L);
124127

125128
// Bucket target options
126129
private static final Storage.BucketTargetOption BUCKET_TARGET_METAGENERATION =
@@ -726,7 +729,10 @@ public void testCreateBlobRetry() throws IOException {
726729
.doReturn(BLOB_INFO1.toPb())
727730
.doThrow(UNEXPECTED_CALL_EXCEPTION)
728731
.when(storageRpcMock)
729-
.create(Mockito.eq(storageObject), capturedStream.capture(), Mockito.eq(EMPTY_RPC_OPTIONS));
732+
.create(
733+
Mockito.eq(storageObject),
734+
capturedStream.capture(),
735+
Mockito.eq(BLOB_INFO1_RPC_OPTIONS_WITH_GENERATION));
730736

731737
storage =
732738
options
@@ -736,7 +742,7 @@ public void testCreateBlobRetry() throws IOException {
736742
.getService();
737743
initializeServiceDependentObjects();
738744

739-
Blob blob = storage.create(BLOB_INFO1, BLOB_CONTENT);
745+
Blob blob = storage.create(BLOB_INFO1, BLOB_CONTENT, BlobTargetOption.generationMatch());
740746

741747
assertEquals(expectedBlob1, blob);
742748

‎google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import com.google.cloud.NoCredentials;
2424
import com.google.cloud.conformance.storage.v1.InstructionList;
2525
import com.google.cloud.conformance.storage.v1.Method;
26-
import com.google.cloud.storage.PackagePrivateMethodWorkarounds;
2726
import com.google.cloud.storage.Storage;
2827
import com.google.cloud.storage.StorageOptions;
2928
import com.google.cloud.storage.conformance.retry.TestBench.RetryTestResource;
@@ -141,7 +140,6 @@ private Storage newStorage(boolean forTest) {
141140
.setHost(testBench.getBaseUri())
142141
.setCredentials(NoCredentials.getInstance())
143142
.setProjectId(testRetryConformance.getProjectId());
144-
builder = PackagePrivateMethodWorkarounds.useDefaultStorageRetryStrategy(builder);
145143
RetrySettings.Builder retrySettingsBuilder =
146144
StorageOptions.getDefaultRetrySettings().toBuilder();
147145
if (forTest) {

0 commit comments

Comments
 (0)