[Dhriti07]

12/n of fast open BiDi reads

Storage.blobReadSession(BlobId, ReadProjectionConfig, BlobSourceOption...) is replaced by blobReadSession(BlobId, List>, BlobSourceOption...), so several reads can be sent with the open. The first readAs with an equal config is given its fast open read, in any order. An empty list only opens the session; a null list is rejected.

The rules are the same as for readSession: at most 100 configs, only asFutureBytes, asFutureByteString and asChannel, no two equal, and asChannel only on its own. The javadoc also says that a fast open read that is never asked for keeps its bytes in memory until the session is closed.

OtelStorageDecorator passes the list through and sets gcs.bidi.fast_open when the list is not empty.

[gemini-code-assist]

Code Review

This pull request updates the blobReadSession API to accept a list of ReadProjectionConfig objects (fastOpenConfigs) instead of a single config, enabling multiple fast-open reads in a single session-opening request. The implementation is updated across GrpcStorageImpl, OtelStorageDecorator, and the Storage interface, along with corresponding test coverage. The review feedback suggests copying the fastOpenConfigs list to an ImmutableList upfront in GrpcStorageImpl to prevent potential NullPointerExceptions from null elements and to avoid copying the list multiple times.

[gemini-code-assist]

To prevent a generic NullPointerException when fastOpenConfigs contains null elements (which would occur when calling c.getType() inside the stream), and to avoid copying the list twice, we can copy fastOpenConfigs to an ImmutableList upfront. This ensures that any null elements are immediately caught with a clean NullPointerException during the copy, and we can reuse the copied list when calling storageDataClient.readSession later.

Suggested change

	requireNonNull(fastOpenConfigs, "fastOpenConfigs must be non null");
	checkArgument(
	fastOpenConfig.getType() == ProjectionType.STREAM_READ,
	"fastOpenConfig must be asFutureBytes(), asFutureByteString() or asChannel(), but was %s",
	fastOpenConfig);
	return blobReadSession(id, ImmutableList.of(fastOpenConfig), options);
	}
	private ApiFuture<BlobReadSession> blobReadSession(
	BlobId id, List<ReadProjectionConfig<?>> fastOpenConfigs, BlobSourceOption... options) {
	fastOpenConfigs.stream().allMatch(c -> c.getType() == ProjectionType.STREAM_READ),
	"fastOpenConfigs must only contain asFutureBytes(), asFutureByteString() or asChannel(),"
	+ " but was %s",
	fastOpenConfigs);
	ImmutableList<ReadProjectionConfig<?>> configs = ImmutableList.copyOf(fastOpenConfigs);
	checkArgument(
	configs.stream().allMatch(c -> c.getType() == ProjectionType.STREAM_READ),
	"fastOpenConfigs must only contain asFutureBytes(), asFutureByteString() or asChannel(),"
	+ " but was %s",
	configs);

References

1. Avoid adding defensive null checks for values that are guaranteed to be non-null by design, as this can hide invariant breaks.

[gemini-code-assist]

We can reuse the pre-copied configs list here instead of copying it again with ImmutableList.copyOf(fastOpenConfigs).

Suggested change

	storageDataClient.readSession(req, context, ImmutableList.copyOf(fastOpenConfigs));
	storageDataClient.readSession(req, context, configs);