Skip to content
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.

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

fix(bigquery/storage/managedwriter): address possible deadlocks #8507

Merged
merged 3 commits into from
Aug 29, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
14 changes: 8 additions & 6 deletions bigquery/storage/managedwriter/managed_stream.go
Expand Up @@ -197,10 +197,11 @@ func (ms *ManagedStream) Finalize(ctx context.Context, opts ...gax.CallOption) (
func (ms *ManagedStream) appendWithRetry(pw *pendingWrite, opts ...gax.CallOption) error {
for {
ms.mu.Lock()
if ms.err != nil {
return ms.err
}
err := ms.err
ms.mu.Unlock()
if err != nil {
return err
}
conn, err := ms.pool.selectConn(pw)
if err != nil {
pw.markDone(nil, err)
Expand Down Expand Up @@ -291,10 +292,11 @@ func (ms *ManagedStream) buildRequest(data [][]byte) *storagepb.AppendRowsReques
func (ms *ManagedStream) AppendRows(ctx context.Context, data [][]byte, opts ...AppendOption) (*AppendResult, error) {
// before we do anything, ensure the writer isn't closed.
ms.mu.Lock()
if ms.err != nil {
return nil, ms.err
}
err := ms.err
ms.mu.Unlock()
if err != nil {
return nil, err
}
// Ensure we build the request and pending write with a consistent schema version.
curSchemaVersion := ms.curDescVersion
req := ms.buildRequest(data)
Expand Down
11 changes: 11 additions & 0 deletions bigquery/storage/managedwriter/managed_stream_test.go
Expand Up @@ -421,6 +421,17 @@ func TestManagedStream_AppendDeadlocks(t *testing.T) {
// Issue two closes, to ensure we're not deadlocking there either.
ms.Close()
ms.Close()

// Issue two more appends, ensure we're not deadlocked as the writer is closed.
gotErr = ms.appendWithRetry(pw)
if !errors.Is(gotErr, io.EOF) {
t.Errorf("expected io.EOF, got %v", gotErr)
}
gotErr = ms.appendWithRetry(pw)
if !errors.Is(gotErr, io.EOF) {
t.Errorf("expected io.EOF, got %v", gotErr)
}

}

}
Expand Down