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.

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

fix: a possible race condition that we used table schema out of the lock. #1575

Merged
merged 1 commit into from
Mar 14, 2022
Merged

fix: a possible race condition that we used table schema out of the lock. #1575

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
46 changes: 13 additions & 33 deletions ...-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,41 +134,21 @@ public ApiFuture<AppendRowsResponse> append(JSONArray jsonArr, long offset)
// Create a new underlying StreamWriter with the updated TableSchema and Descriptor
this.streamWriter = streamWriterBuilder.setWriterSchema(this.protoSchema).build();
}
}

ProtoRows.Builder rowsBuilder = ProtoRows.newBuilder();
// Any error in convertJsonToProtoMessage will throw an
// IllegalArgumentException/IllegalStateException/NullPointerException and will halt processing
// of JSON data.
long currentRequestSize = 0;
for (int i = 0; i < jsonArr.length(); i++) {
JSONObject json = jsonArr.getJSONObject(i);
Message protoMessage =
JsonToProtoMessage.convertJsonToProtoMessage(
this.descriptor, this.tableSchema, json, ignoreUnknownFields);
rowsBuilder.addSerializedRows(protoMessage.toByteString());
currentRequestSize += protoMessage.getSerializedSize();
}
// Need to make sure refreshAppendAndSetDescriptor finish first before this can run
synchronized (this) {
this.totalMessageSize += currentRequestSize;
this.absTotal += currentRequestSize;
// Reconnect on every 9.5MB.
if (this.totalMessageSize > 9500000 && this.reconnectAfter10M) {
streamWriter.close();
// Create a new underlying StreamWriter aka establish a new connection.
this.streamWriter = streamWriterBuilder.setWriterSchema(protoSchema).build();
this.totalMessageSize = this.protoSchema.getSerializedSize() + currentRequestSize;
this.absTotal += currentRequestSize;
// Allow first request to pass.
ProtoRows.Builder rowsBuilder = ProtoRows.newBuilder();
// Any error in convertJsonToProtoMessage will throw an
// IllegalArgumentException/IllegalStateException/NullPointerException and will halt
// processing
// of JSON data.
long currentRequestSize = 0;
for (int i = 0; i < jsonArr.length(); i++) {
JSONObject json = jsonArr.getJSONObject(i);
Message protoMessage =
JsonToProtoMessage.convertJsonToProtoMessage(
this.descriptor, this.tableSchema, json, ignoreUnknownFields);
rowsBuilder.addSerializedRows(protoMessage.toByteString());
currentRequestSize += protoMessage.getSerializedSize();
}
LOG.fine(
"Sending a total of:"
+ this.totalMessageSize
+ " "
+ currentRequestSize
+ " "
+ this.absTotal);
final ApiFuture<AppendRowsResponse> appendResponseFuture =
this.streamWriter.append(rowsBuilder.build(), offset);
return appendResponseFuture;
Expand Down