[Dhriti07]

2/n of fast open BiDi reads.

ObjectReadSessionImpl keeps the fast open reads sent with the open, keyed by the config each was created from. readAs removes the read for an equal config and returns it, so the bytes that came back with the open are used and no second request is sent; any other readAs sends a new read exactly as before. Closing the session also closes fast open reads that were never asked for.

[gemini-code-assist]

Code Review

This pull request implements a fast-open read mechanism for Google Cloud Storage sessions, allowing reads to be sent alongside the initial session-opening request to save a round trip. The session caches these fast-open reads and returns them on the first matching readAs call, while ensuring any unclaimed reads are properly closed. The review feedback highlights a critical issue where duplicate configurations in the fast-open list would overwrite each other in the map, leading to resource leaks and redundant requests. To resolve this, the reviewer suggests refactoring the map to store a queue of reads per configuration, providing complete code suggestions across the session implementation and client code.

[gemini-code-assist]

If fastOpenConfigs contains duplicate configs (e.g., requesting the same range multiple times), using a simple Map where the key is the config will cause subsequent duplicate configs to overwrite previous ones. This results in only the last fast-open read being preserved, while the others are leaked/wasted and subsequent readAs calls for those configs will unnecessarily trigger new requests.

To support duplicate configs correctly and safely under concurrent access, we can map each config to a thread-safe queue (such as ConcurrentLinkedQueue) of outstanding reads.

Suggested change

	// Fast open reads: reads sent in the request that opened the stream, keyed by the config each was
	// created from. The first readAs with an equal config is given the read instead of sending a new
	// one.
	private final ConcurrentMap<ReadProjectionConfig<?>, ObjectReadSessionStreamRead<?>>
	fastOpenReads;
	// Fast open reads: reads sent in the request that opened the stream, keyed by the config each was
	// created from. The first readAs with an equal config is given the read instead of sending a new
	// one. Multiple identical configs are supported via a queue.
	private final ConcurrentMap<ReadProjectionConfig<?>, java.util.Queue<ObjectReadSessionStreamRead<?>>>
	fastOpenReads;

[gemini-code-assist]

Update the constructor to accept the queue-based map of fast-open reads.

Suggested change

	ObjectReadSessionImpl(
	ScheduledExecutorService executor,
	ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,
	ObjectReadSessionStream stream,
	ObjectReadSessionState state,
	RetryContextProvider retryContextProvider) {
	RetryContextProvider retryContextProvider,
	Map<ReadProjectionConfig<?>, ObjectReadSessionStreamRead<?>> fastOpenReads) {
	this.executor = executor;
	this.callable = callable;
	this.stream = stream;
	this.state = state;
	this.resource = state.getMetadata();
	this.retryContextProvider = retryContextProvider;
	this.children = new ConcurrentIdentityMap<>();
	this.fastOpenReads = new ConcurrentHashMap<>(fastOpenReads);
	this.open = true;
	}
	ObjectReadSessionImpl(
	ScheduledExecutorService executor,
	ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,
	ObjectReadSessionStream stream,
	ObjectReadSessionState state,
	RetryContextProvider retryContextProvider,
	Map<ReadProjectionConfig<?>, java.util.Queue<ObjectReadSessionStreamRead<?>>> fastOpenReads) {
	this.executor = executor;
	this.callable = callable;
	this.stream = stream;
	this.state = state;
	this.resource = state.getMetadata();
	this.retryContextProvider = retryContextProvider;
	this.children = new ConcurrentIdentityMap<>();
	this.fastOpenReads = new ConcurrentHashMap<>(fastOpenReads);
	this.open = true;
	}

[gemini-code-assist]

Update readAs to poll from the queue of fast-open reads for the given config.

Suggested change

	ObjectReadSessionStreamRead<?> fastOpenRead = fastOpenReads.remove(config);
	if (fastOpenRead != null) {
	// configs are final classes with a class-exact equals, so an equal config has the same
	// Projection type
	@SuppressWarnings("unchecked")
	Projection projection = (Projection) fastOpenRead.project();
	return projection;
	}
	java.util.Queue<ObjectReadSessionStreamRead<?>> queue = fastOpenReads.get(config);
	ObjectReadSessionStreamRead<?> fastOpenRead = queue != null ? queue.poll() : null;
	if (fastOpenRead != null) {
	// configs are final classes with a class-exact equals, so an equal config has the same
	// Projection type
	@SuppressWarnings("unchecked")
	Projection projection = (Projection) fastOpenRead.project();
	return projection;
	}

[gemini-code-assist]

Update the finally block in close() to close all remaining fast-open reads in the queues.

Suggested change

	} finally {
	// a fast open read nobody asked for may already be complete and out of the state, so
	// stream.close() above does not reach it; close it here to release its buffers
	GrpcUtils.closeAll(fastOpenReads.values());
	fastOpenReads.clear();
	}
	} finally {
	// a fast open read nobody asked for may already be complete and out of the state, so
	// stream.close() above does not reach it; close it here to release its buffers
	fastOpenReads.values().forEach(GrpcUtils::closeAll);
	fastOpenReads.clear();
	}

[gemini-code-assist]

Update readSession to group the fast-open reads by config into a queue to support duplicate configs.

Suggested change

	Map<ReadProjectionConfig<?>, ObjectReadSessionStreamRead<?>> fastOpenReads = new HashMap<>();
	for (ReadProjectionConfig<?> config : fastOpenConfigs) {
	long readId = state.newReadId();
	ObjectReadSessionStreamRead<?> read =
	config.cast().newRead(readId, retryContextProvider.create());
	state.putOutstandingRead(readId, read);
	fastOpenReads.put(config, read);
	}
	Map<ReadProjectionConfig<?>, java.util.Queue<ObjectReadSessionStreamRead<?>>> fastOpenReads = new HashMap<>();
	for (ReadProjectionConfig<?> config : fastOpenConfigs) {
	long readId = state.newReadId();
	ObjectReadSessionStreamRead<?> read =
	config.cast().newRead(readId, retryContextProvider.create());
	state.putOutstandingRead(readId, read);
	fastOpenReads.computeIfAbsent(config, k -> new java.util.concurrent.ConcurrentLinkedQueue<>()).add(read);
	}