Skip to content

fix(storage): fix flaky storage integration tests #13253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nidhiii-27 wants to merge 4 commits into from
Closed

fix(storage): fix flaky storage integration tests #13253

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6ea8379
fix(storage): fix flaky storage integration tests
nidhiii-27 May 22, 2026
cef75cd
add terminal_error check
nidhiii-27 May 22, 2026
44caa5f
change worker-class in docker starting the testbench
nidhiii-27 May 22, 2026
bd99c3d
try changing the connection URI
nidhiii-27 May 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions ...oogle-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,9 @@ public void flush() {
public boolean finishWrite(long length) {
lock.lock();
try {
if (state.getState() == State.TERMINAL_SUCCESS) {
return true;
}

@gemini-code-assist gemini-code-assist Bot May 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The check only accounts for State.TERMINAL_SUCCESS. If the stream has reached a terminal failure state (e.g., TERMINAL_FAILURE), subsequent calls to finishWrite might still trigger an IllegalStateException when calling state.isFinalizing() or other methods that enforce state transitions. Consider checking for any terminal state to ensure the method is robust against all completed stream states.

// if we're already finalizing, ack rather than enqueueing again
if (state.isFinalizing() && state.getTotalSentBytes() == length) {
return true;
Expand Down Expand Up @@ -192,6 +195,10 @@ public boolean closeStream(long length) {
lock.lock();
try {

if (state.getState() == State.TERMINAL_SUCCESS) {
return true;
}

@gemini-code-assist gemini-code-assist Bot May 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similar to the change in finishWrite, this check only handles TERMINAL_SUCCESS. To prevent IllegalStateException when the stream has failed or been cancelled, it would be safer to check for any terminal state before proceeding with state.finalFlush(length).


boolean offer = state.finalFlush(length);
if (offer) {
internalSend();
Expand Down
5 changes: 5 additions & 0 deletions java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/BidiUploadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1356,6 +1356,11 @@ boolean isFinalizing() {
return false;
}

@Override
State getState() {
return State.RUNNING;
}

@Override
long getTotalSentBytes() {
return 0;
Expand Down
7 changes: 3 additions & 4 deletions ...rage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -380,12 +380,9 @@ static class DirectWriteService extends StorageImplBase {
private static final Logger LOGGER = LoggerFactory.getLogger(DirectWriteService.class);
private final BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c;

private ImmutableList.Builder<WriteObjectRequest> requests;

DirectWriteService(
BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c) {
this.c = c;
this.requests = new ImmutableList.Builder<>();
}

DirectWriteService(ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes) {
Expand Down Expand Up @@ -420,6 +417,9 @@ private static void logUnexpectedRequest(
@Override
public StreamObserver<WriteObjectRequest> writeObject(StreamObserver<WriteObjectResponse> obs) {
return new Adapter() {
private final ImmutableList.Builder<WriteObjectRequest> requests =
new ImmutableList.Builder<>();

@Override
public void onNext(WriteObjectRequest value) {
requests.add(value);
Expand All @@ -432,7 +432,6 @@ public void onError(Throwable t) {}
public void onCompleted() {
ImmutableList<WriteObjectRequest> build = requests.build();
c.accept(obs, build);
requests = new ImmutableList.Builder<>();
}
};
}
Expand Down
Loading