Skip to content

Commit c069aca

Browse files
BenWhiteheadBenWhitehead
authored
chore: blob write session cleanup (#2280)
1. Add TransportCompatibility annotations to factory methods in BlobWriteSessionConfigs 2. Add BlobWriteSessionConfig.GrpcCompatible marker interface for validation in GrpcStorageOptions 3. Add BlobWriteSessionConfig.GrpcCompatible to all existing BlobWriteSessionConfigs 4. Cleanup todo in DefaultBlobWriteSessionConfig that was reaching down into the GrpcBlobWriteChannel. Now, the WritableByteChannelSession will be constructed directly. 5. Add @CrossRun annotations to BlobWriteSession integration tests
1 parent d3b7bb3 commit c069aca

12 files changed

+158
-101
lines changed

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

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,9 @@
1717
package com.google.cloud.storage;
1818

1919
import com.google.api.core.InternalApi;
20-
import com.google.cloud.storage.Conversions.Decoder;
2120
import com.google.cloud.storage.Storage.BlobWriteOption;
2221
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
2322
import com.google.cloud.storage.UnifiedOpts.Opts;
24-
import com.google.storage.v2.WriteObjectResponse;
2523
import java.io.IOException;
2624
import java.io.Serializable;
2725
import java.time.Clock;
@@ -52,9 +50,15 @@ public abstract class BlobWriteSessionConfig implements Serializable {
5250
interface WriterFactory {
5351
@InternalApi
5452
WritableByteChannelSession<?, BlobInfo> writeSession(
55-
StorageInternal s,
56-
BlobInfo info,
57-
Opts<ObjectTargetOpt> opts,
58-
Decoder<WriteObjectResponse, BlobInfo> d);
53+
StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts);
5954
}
55+
56+
/**
57+
* Internal marker interface to signify an implementation of {@link BlobWriteSessionConfig} is
58+
* compatible with {@link com.google.cloud.storage.TransportCompatibility.Transport#GRPC}
59+
*
60+
* <p>We could evaluate the annotations, but the code for that is more complicated and probably
61+
* not worth the effort.
62+
*/
63+
interface GrpcCompatible {}
6064
}

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
2222
import com.google.cloud.storage.Storage.BlobTargetOption;
2323
import com.google.cloud.storage.Storage.BlobWriteOption;
24+
import com.google.cloud.storage.TransportCompatibility.Transport;
2425
import com.google.common.collect.ImmutableList;
2526
import java.io.IOException;
2627
import java.nio.channels.WritableByteChannel;
@@ -240,6 +241,7 @@ private BlobWriteSessionConfigs() {}
240241
* @since 2.26.0 This new api is in preview and is subject to breaking changes.
241242
*/
242243
@BetaApi
244+
@TransportCompatibility({Transport.GRPC})
243245
public static DefaultBlobWriteSessionConfig getDefault() {
244246
return new DefaultBlobWriteSessionConfig(ByteSizeConstants._16MiB);
245247
}
@@ -255,6 +257,7 @@ public static DefaultBlobWriteSessionConfig getDefault() {
255257
* @since 2.26.0 This new api is in preview and is subject to breaking changes.
256258
*/
257259
@BetaApi
260+
@TransportCompatibility({Transport.GRPC})
258261
public static BlobWriteSessionConfig bufferToTempDirThenUpload() throws IOException {
259262
return bufferToDiskThenUpload(
260263
Paths.get(System.getProperty("java.io.tmpdir"), "google-cloud-storage"));
@@ -271,6 +274,7 @@ public static BlobWriteSessionConfig bufferToTempDirThenUpload() throws IOExcept
271274
* @since 2.26.0 This new api is in preview and is subject to breaking changes.
272275
*/
273276
@BetaApi
277+
@TransportCompatibility({Transport.GRPC})
274278
public static BufferToDiskThenUpload bufferToDiskThenUpload(Path path) throws IOException {
275279
return bufferToDiskThenUpload(ImmutableList.of(path));
276280
}
@@ -289,6 +293,7 @@ public static BufferToDiskThenUpload bufferToDiskThenUpload(Path path) throws IO
289293
* @since 2.26.0 This new api is in preview and is subject to breaking changes.
290294
*/
291295
@BetaApi
296+
@TransportCompatibility({Transport.GRPC})
292297
public static BufferToDiskThenUpload bufferToDiskThenUpload(Collection<Path> paths)
293298
throws IOException {
294299
return new BufferToDiskThenUpload(ImmutableList.copyOf(paths), false);
@@ -306,6 +311,7 @@ public static BufferToDiskThenUpload bufferToDiskThenUpload(Collection<Path> pat
306311
* @since 2.27.0 This new api is in preview and is subject to breaking changes.
307312
*/
308313
@BetaApi
314+
@TransportCompatibility(Transport.GRPC)
309315
public static JournalingBlobWriteSessionConfig journaling(Collection<Path> paths) {
310316
return new JournalingBlobWriteSessionConfig(ImmutableList.copyOf(paths), false);
311317
}
@@ -321,6 +327,7 @@ public static JournalingBlobWriteSessionConfig journaling(Collection<Path> paths
321327
* @since 2.28.0 This new api is in preview and is subject to breaking changes.
322328
*/
323329
@BetaApi
330+
@TransportCompatibility({Transport.GRPC})
324331
public static ParallelCompositeUploadBlobWriteSessionConfig parallelCompositeUpload() {
325332
return ParallelCompositeUploadBlobWriteSessionConfig.withDefaults();
326333
}

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

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -21,15 +21,14 @@
2121
import com.google.api.core.BetaApi;
2222
import com.google.api.core.InternalApi;
2323
import com.google.api.core.SettableApiFuture;
24-
import com.google.cloud.storage.Conversions.Decoder;
2524
import com.google.cloud.storage.RecoveryFileManager.RecoveryVolumeSinkFactory;
2625
import com.google.cloud.storage.Storage.BlobWriteOption;
26+
import com.google.cloud.storage.TransportCompatibility.Transport;
2727
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
2828
import com.google.cloud.storage.UnifiedOpts.Opts;
2929
import com.google.common.annotations.VisibleForTesting;
3030
import com.google.common.collect.ImmutableList;
3131
import com.google.common.util.concurrent.MoreExecutors;
32-
import com.google.storage.v2.WriteObjectResponse;
3332
import java.io.IOException;
3433
import java.io.ObjectInputStream;
3534
import java.io.ObjectOutputStream;
@@ -60,7 +59,9 @@
6059
*/
6160
@Immutable
6261
@BetaApi
63-
public final class BufferToDiskThenUpload extends BlobWriteSessionConfig {
62+
@TransportCompatibility({Transport.GRPC})
63+
public final class BufferToDiskThenUpload extends BlobWriteSessionConfig
64+
implements BlobWriteSessionConfig.GrpcCompatible {
6465
private static final long serialVersionUID = 9059242302276891867L;
6566

6667
/**
@@ -151,10 +152,7 @@ private Factory(RecoveryFileManager recoveryFileManager, Clock clock, Throughput
151152
@InternalApi
152153
@Override
153154
public WritableByteChannelSession<?, BlobInfo> writeSession(
154-
StorageInternal storage,
155-
BlobInfo info,
156-
Opts<ObjectTargetOpt> opts,
157-
Decoder<WriteObjectResponse, BlobInfo> d) {
155+
StorageInternal storage, BlobInfo info, Opts<ObjectTargetOpt> opts) {
158156
return new Factory.WriteToFileThenUpload(
159157
storage, info, opts, recoveryFileManager.newRecoveryFile(info));
160158
}

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

Lines changed: 56 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,15 @@
2020
import com.google.api.core.ApiFutures;
2121
import com.google.api.core.BetaApi;
2222
import com.google.api.core.InternalApi;
23+
import com.google.api.gax.grpc.GrpcCallContext;
2324
import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
2425
import com.google.cloud.storage.Conversions.Decoder;
26+
import com.google.cloud.storage.TransportCompatibility.Transport;
2527
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
2628
import com.google.cloud.storage.UnifiedOpts.Opts;
2729
import com.google.common.base.Preconditions;
2830
import com.google.common.util.concurrent.MoreExecutors;
31+
import com.google.storage.v2.WriteObjectRequest;
2932
import com.google.storage.v2.WriteObjectResponse;
3033
import java.nio.channels.WritableByteChannel;
3134
import java.time.Clock;
@@ -52,7 +55,9 @@
5255
*/
5356
@Immutable
5457
@BetaApi
55-
public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig {
58+
@TransportCompatibility({Transport.GRPC})
59+
public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
60+
implements BlobWriteSessionConfig.GrpcCompatible {
5661
private static final long serialVersionUID = -6873740918589930633L;
5762

5863
private final int chunkSize;
@@ -102,6 +107,9 @@ WriterFactory createFactory(Clock clock) {
102107

103108
@InternalApi
104109
private static final class Factory implements WriterFactory {
110+
private static final Decoder<WriteObjectResponse, BlobInfo>
111+
WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER =
112+
Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource);
105113

106114
private final int chunkSize;
107115

@@ -112,21 +120,36 @@ private Factory(int chunkSize) {
112120
@InternalApi
113121
@Override
114122
public WritableByteChannelSession<?, BlobInfo> writeSession(
115-
StorageInternal s,
116-
BlobInfo info,
117-
Opts<ObjectTargetOpt> opts,
118-
Decoder<WriteObjectResponse, BlobInfo> d) {
119-
// todo: invert this
120-
// make GrpcBlobWriteChannel use this factory to produce its WriteSession
123+
StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts) {
121124
if (s instanceof GrpcStorageImpl) {
122-
GrpcStorageImpl g = (GrpcStorageImpl) s;
123-
GrpcBlobWriteChannel writer = g.internalWriter(info, opts);
124-
writer.setChunkSize(chunkSize);
125-
WritableByteChannelSession<BufferedWritableByteChannel, WriteObjectResponse> session =
126-
writer.newLazyWriteChannel().getSession();
127-
return new DecoratedWritableByteChannelSession<>(session, d);
125+
return new DecoratedWritableByteChannelSession<>(
126+
new LazySession<>(
127+
new LazyWriteChannel<>(
128+
() -> {
129+
GrpcStorageImpl grpc = (GrpcStorageImpl) s;
130+
GrpcCallContext grpcCallContext =
131+
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
132+
WriteObjectRequest req = grpc.getWriteObjectRequest(info, opts);
133+
134+
ApiFuture<ResumableWrite> startResumableWrite =
135+
grpc.startResumableWrite(grpcCallContext, req);
136+
return ResumableMedia.gapic()
137+
.write()
138+
.byteChannel(grpc.storageClient.writeObjectCallable())
139+
.setHasher(Hasher.noop())
140+
.setByteStringStrategy(ByteStringStrategy.copy())
141+
.resumable()
142+
.withRetryConfig(
143+
grpc.getOptions(), grpc.retryAlgorithmManager.idempotent())
144+
.buffered(BufferHandle.allocate(chunkSize))
145+
.setStartAsync(startResumableWrite)
146+
.build();
147+
})),
148+
WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
149+
} else {
150+
throw new IllegalStateException(
151+
"Unknown Storage implementation: " + s.getClass().getName());
128152
}
129-
return CrossTransportUtils.throwGrpcOnly(DefaultBlobWriteSessionConfig.class, "");
130153
}
131154
}
132155

@@ -162,4 +185,23 @@ public ApiFuture<BlobInfo> getResult() {
162185
delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
163186
}
164187
}
188+
189+
private static final class LazySession<R>
190+
implements WritableByteChannelSession<BufferedWritableByteChannel, R> {
191+
private final LazyWriteChannel<R> lazy;
192+
193+
private LazySession(LazyWriteChannel<R> lazy) {
194+
this.lazy = lazy;
195+
}
196+
197+
@Override
198+
public ApiFuture<BufferedWritableByteChannel> openAsync() {
199+
return lazy.getSession().openAsync();
200+
}
201+
202+
@Override
203+
public ApiFuture<R> getResult() {
204+
return lazy.getSession().getResult();
205+
}
206+
}
165207
}

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

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
176176
final GrpcConversions codecs;
177177
final GrpcRetryAlgorithmManager retryAlgorithmManager;
178178
final SyntaxDecoders syntaxDecoders;
179-
private final Decoder<WriteObjectResponse, BlobInfo> writeObjectResponseBlobInfoDecoder;
180179

181180
// workaround for https://github.com/googleapis/java-storage/issues/1736
182181
private final Opts<UserProject> defaultOpts;
@@ -194,8 +193,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
194193
this.codecs = Conversions.grpc();
195194
this.retryAlgorithmManager = options.getRetryAlgorithmManager();
196195
this.syntaxDecoders = new SyntaxDecoders();
197-
this.writeObjectResponseBlobInfoDecoder =
198-
codecs.blobInfo().compose(WriteObjectResponse::getResource);
199196
this.defaultProjectId = UnifiedOpts.projectId(options.getProjectId());
200197
}
201198

@@ -719,14 +716,9 @@ public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption.
719716
@Override
720717
public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) {
721718
Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts);
722-
return internalWriter(blobInfo, opts);
723-
}
724-
725-
@Override
726-
public GrpcBlobWriteChannel internalWriter(BlobInfo info, Opts<ObjectTargetOpt> opts) {
727719
GrpcCallContext grpcCallContext =
728720
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
729-
WriteObjectRequest req = getWriteObjectRequest(info, opts);
721+
WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
730722
Hasher hasher = Hasher.noop();
731723
return new GrpcBlobWriteChannel(
732724
storageClient.writeObjectCallable(),
@@ -1534,7 +1526,7 @@ public boolean deleteNotification(String bucket, String notificationId) {
15341526
public BlobWriteSession blobWriteSession(BlobInfo info, BlobWriteOption... options) {
15351527
Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(info);
15361528
WritableByteChannelSession<?, BlobInfo> writableByteChannelSession =
1537-
writerFactory.writeSession(this, info, opts, writeObjectResponseBlobInfoDecoder);
1529+
writerFactory.writeSession(this, info, opts);
15381530
return BlobWriteSessions.of(writableByteChannelSession);
15391531
}
15401532

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package com.google.cloud.storage;
1818

19+
import static com.google.common.base.Preconditions.checkArgument;
1920
import static java.util.Objects.requireNonNull;
2021

2122
import com.google.api.core.ApiClock;
@@ -533,6 +534,9 @@ public GrpcStorageOptions.Builder setGrpcInterceptorProvider(
533534
public GrpcStorageOptions.Builder setBlobWriteSessionConfig(
534535
@NonNull BlobWriteSessionConfig blobWriteSessionConfig) {
535536
requireNonNull(blobWriteSessionConfig, "blobWriteSessionConfig must be non null");
537+
checkArgument(
538+
blobWriteSessionConfig instanceof BlobWriteSessionConfig.GrpcCompatible,
539+
"The provided instance of BlobWriteSessionConfig is not compatible with gRPC transport.");
536540
this.blobWriteSessionConfig = blobWriteSessionConfig;
537541
return this;
538542
}

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

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,8 @@
6868
@Immutable
6969
@BetaApi
7070
@TransportCompatibility(Transport.GRPC)
71-
public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConfig {
71+
public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConfig
72+
implements BlobWriteSessionConfig.GrpcCompatible {
7273
private static final long serialVersionUID = 9059242302276891867L;
7374

7475
/**
@@ -159,10 +160,7 @@ private Factory(RecoveryFileManager recoveryFileManager, Clock clock, Throughput
159160
@InternalApi
160161
@Override
161162
public WritableByteChannelSession<?, BlobInfo> writeSession(
162-
StorageInternal storage,
163-
BlobInfo info,
164-
Opts<ObjectTargetOpt> opts,
165-
Decoder<WriteObjectResponse, BlobInfo> d) {
163+
StorageInternal storage, BlobInfo info, Opts<ObjectTargetOpt> opts) {
166164
if (storage instanceof GrpcStorageImpl) {
167165
GrpcStorageImpl grpcStorage = (GrpcStorageImpl) storage;
168166
RecoveryFile recoveryFile = recoveryFileManager.newRecoveryFile(info);
@@ -189,7 +187,7 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
189187
.setStartAsync(start)
190188
.build();
191189

192-
return new JournalingUpload<>(session, start, d);
190+
return new JournalingUpload<>(session, start);
193191
} else {
194192
return CrossTransportUtils.throwHttpJsonOnly(BlobWriteSessionConfigs.class, "journaling");
195193
}
@@ -204,11 +202,10 @@ private final class JournalingUpload<WBC extends WritableByteChannel>
204202

205203
public JournalingUpload(
206204
WritableByteChannelSession<WBC, WriteObjectResponse> session,
207-
ApiFuture<WriteCtx<ResumableWrite>> start,
208-
Decoder<WriteObjectResponse, BlobInfo> decoder) {
205+
ApiFuture<WriteCtx<ResumableWrite>> start) {
209206
this.session = session;
210207
this.start = start;
211-
this.decoder = decoder;
208+
this.decoder = Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource);
212209
}
213210

214211
@Override

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

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -26,17 +26,16 @@
2626
import com.google.api.core.InternalApi;
2727
import com.google.api.core.SettableApiFuture;
2828
import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
29-
import com.google.cloud.storage.Conversions.Decoder;
3029
import com.google.cloud.storage.MetadataField.PartRange;
3130
import com.google.cloud.storage.Storage.BlobWriteOption;
31+
import com.google.cloud.storage.TransportCompatibility.Transport;
3232
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
3333
import com.google.cloud.storage.UnifiedOpts.Opts;
3434
import com.google.common.annotations.VisibleForTesting;
3535
import com.google.common.hash.HashCode;
3636
import com.google.common.hash.HashFunction;
3737
import com.google.common.hash.Hashing;
3838
import com.google.common.util.concurrent.ThreadFactoryBuilder;
39-
import com.google.storage.v2.WriteObjectResponse;
4039
import java.io.IOException;
4140
import java.io.ObjectOutputStream;
4241
import java.io.Serializable;
@@ -116,7 +115,9 @@
116115
*/
117116
@Immutable
118117
@BetaApi
119-
public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessionConfig {
118+
@TransportCompatibility({Transport.GRPC})
119+
public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessionConfig
120+
implements BlobWriteSessionConfig.GrpcCompatible {
120121

121122
private static final int MAX_PARTS_PER_COMPOSE = 32;
122123
private final int maxPartsPerCompose;
@@ -669,10 +670,7 @@ private ParallelCompositeUploadWriterFactory(
669670

670671
@Override
671672
public WritableByteChannelSession<?, BlobInfo> writeSession(
672-
StorageInternal s,
673-
BlobInfo info,
674-
Opts<ObjectTargetOpt> opts,
675-
Decoder<WriteObjectResponse, BlobInfo> d) {
673+
StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts) {
676674
return new PCUSession(s, info, opts);
677675
}
678676

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

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,6 @@ default BlobInfo internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetO
3131
throw new UnsupportedOperationException("not implemented");
3232
}
3333

34-
default StorageWriteChannel internalWriter(BlobInfo info, Opts<ObjectTargetOpt> opts) {
35-
throw new UnsupportedOperationException("not implemented");
36-
}
37-
3834
default BlobInfo internalDirectUpload(BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
3935
throw new UnsupportedOperationException("not implemented");
4036
}

0 commit comments

Comments
 (0)