Skip to content

Commit 5b000d0

Browse files
egreco12Evan Grecogcf-owl-bot[bot]
authored
feat: Add RetrySettings use to Write API samples. (#2419)
--------- Co-authored-by: Evan Greco <egreco@google.com> Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
1 parent 4d5eb73 commit 5b000d0

File tree

7 files changed

+102
-43
lines changed

7 files changed

+102
-43
lines changed

‎google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriter.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -362,7 +362,7 @@ public Builder setCompressorName(String compressorName) {
362362
* Enable client lib automatic retries on request level errors.
363363
*
364364
* <pre>
365-
* Immeidate Retry code:
365+
* Immediate Retry code:
366366
* ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
367367
* Backoff Retry code:
368368
* RESOURCE_EXHAUSTED

‎samples/snippets/src/main/java/com/example/bigquerystorage/JsonWriterStreamCdc.java

+17-1
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import com.google.api.core.ApiFuture;
2020
import com.google.api.core.ApiFutureCallback;
2121
import com.google.api.core.ApiFutures;
22+
import com.google.api.gax.retrying.RetrySettings;
2223
import com.google.cloud.bigquery.BigQuery;
2324
import com.google.cloud.bigquery.BigQueryException;
2425
import com.google.cloud.bigquery.BigQueryOptions;
@@ -37,6 +38,7 @@
3738
import java.io.IOException;
3839
import org.json.JSONArray;
3940
import org.json.JSONObject;
41+
import org.threeten.bp.Duration;
4042

4143
public class JsonWriterStreamCdc {
4244

@@ -108,6 +110,18 @@ private static void query(String query) {
108110
public static void writeToDefaultStream(
109111
String projectId, String datasetName, String tableName, JSONArray data)
110112
throws DescriptorValidationException, InterruptedException, IOException {
113+
// Configure in-stream automatic retry settings.
114+
// Error codes that are immediately retried:
115+
// * ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
116+
// Error codes that are retried with exponential backoff:
117+
// * RESOURCE_EXHAUSTED
118+
RetrySettings retrySettings =
119+
RetrySettings.newBuilder()
120+
.setInitialRetryDelay(Duration.ofMillis(500))
121+
.setRetryDelayMultiplier(1.1)
122+
.setMaxAttempts(5)
123+
.setMaxRetryDelay(Duration.ofMinutes(1))
124+
.build();
111125
// To use the UPSERT functionality, the table schema needs to be padded with an additional
112126
// column "_change_type".
113127
TableSchema tableSchema =
@@ -159,7 +173,9 @@ public static void writeToDefaultStream(
159173
// Use the JSON stream writer to send records in JSON format.
160174
TableName parentTable = TableName.of(projectId, datasetName, tableName);
161175
try (JsonStreamWriter writer =
162-
JsonStreamWriter.newBuilder(parentTable.toString(), tableSchema).build()) {
176+
JsonStreamWriter.newBuilder(parentTable.toString(), tableSchema)
177+
.setRetrySettings(retrySettings)
178+
.build()) {
163179

164180
ApiFuture<AppendRowsResponse> future = writer.append(data);
165181
// The append method is asynchronous. Rather than waiting for the method to complete,

‎samples/snippets/src/main/java/com/example/bigquerystorage/ParallelWriteCommittedStream.java

+15-1
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import com.google.api.core.ApiFuture;
2121
import com.google.api.core.ApiFutureCallback;
2222
import com.google.api.core.ApiFutures;
23+
import com.google.api.gax.retrying.RetrySettings;
2324
import com.google.cloud.bigquery.storage.v1.AppendRowsResponse;
2425
import com.google.cloud.bigquery.storage.v1.BQTableSchemaToProtoDescriptor;
2526
import com.google.cloud.bigquery.storage.v1.BigQueryWriteClient;
@@ -36,12 +37,12 @@
3637
import com.google.protobuf.Descriptors.DescriptorValidationException;
3738
import com.google.protobuf.Message;
3839
import java.io.IOException;
39-
import java.time.Duration;
4040
import java.util.concurrent.ThreadLocalRandom;
4141
import java.util.logging.Logger;
4242
import javax.annotation.Nullable;
4343
import javax.annotation.concurrent.GuardedBy;
4444
import org.json.JSONObject;
45+
import org.threeten.bp.Duration;
4546

4647
public class ParallelWriteCommittedStream {
4748

@@ -157,13 +158,26 @@ private void writeToStream(
157158
lastMetricsSuccessCount = 0;
158159
lastMetricsFailureCount = 0;
159160
}
161+
// Configure in-stream automatic retry settings.
162+
// Error codes that are immediately retried:
163+
// * ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
164+
// Error codes that are retried with exponential backoff:
165+
// * RESOURCE_EXHAUSTED
166+
RetrySettings retrySettings =
167+
RetrySettings.newBuilder()
168+
.setInitialRetryDelay(Duration.ofMillis(500))
169+
.setRetryDelayMultiplier(1.1)
170+
.setMaxAttempts(5)
171+
.setMaxRetryDelay(Duration.ofMinutes(1))
172+
.build();
160173
Descriptor descriptor =
161174
BQTableSchemaToProtoDescriptor.convertBQTableSchemaToProtoDescriptor(
162175
writeStream.getTableSchema());
163176
ProtoSchema protoSchema = ProtoSchemaConverter.convert(descriptor);
164177
try (StreamWriter writer =
165178
StreamWriter.newBuilder(writeStream.getName())
166179
.setWriterSchema(protoSchema)
180+
.setRetrySettings(retrySettings)
167181
.setTraceId("SAMPLE:parallel_append")
168182
.build()) {
169183
while (System.currentTimeMillis() < deadlineMillis) {

‎samples/snippets/src/main/java/com/example/bigquerystorage/WriteBufferedStream.java

+16
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
// [START bigquerystorage_jsonstreamwriter_buffered]
2020
import com.google.api.core.ApiFuture;
21+
import com.google.api.gax.retrying.RetrySettings;
2122
import com.google.cloud.bigquery.storage.v1.AppendRowsResponse;
2223
import com.google.cloud.bigquery.storage.v1.BigQueryWriteClient;
2324
import com.google.cloud.bigquery.storage.v1.CreateWriteStreamRequest;
@@ -33,6 +34,7 @@
3334
import java.util.concurrent.ExecutionException;
3435
import org.json.JSONArray;
3536
import org.json.JSONObject;
37+
import org.threeten.bp.Duration;
3638

3739
public class WriteBufferedStream {
3840

@@ -61,11 +63,25 @@ public static void writeBufferedStream(String projectId, String datasetName, Str
6163
.build();
6264
WriteStream writeStream = client.createWriteStream(createWriteStreamRequest);
6365

66+
// Configure in-stream automatic retry settings.
67+
// Error codes that are immediately retried:
68+
// * ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
69+
// Error codes that are retried with exponential backoff:
70+
// * RESOURCE_EXHAUSTED
71+
RetrySettings retrySettings =
72+
RetrySettings.newBuilder()
73+
.setInitialRetryDelay(Duration.ofMillis(500))
74+
.setRetryDelayMultiplier(1.1)
75+
.setMaxAttempts(5)
76+
.setMaxRetryDelay(Duration.ofMinutes(1))
77+
.build();
78+
6479
// Use the JSON stream writer to send records in JSON format.
6580
// For more information about JsonStreamWriter, see:
6681
// https://googleapis.dev/java/google-cloud-bigquerystorage/latest/com/google/cloud/bigquery/storage/v1beta2/JsonStreamWriter.html
6782
try (JsonStreamWriter writer =
6883
JsonStreamWriter.newBuilder(writeStream.getName(), writeStream.getTableSchema())
84+
.setRetrySettings(retrySettings)
6985
.build()) {
7086
// Write two batches to the stream, each with 10 JSON records.
7187
for (int i = 0; i < 2; i++) {

‎samples/snippets/src/main/java/com/example/bigquerystorage/WriteCommittedStream.java

+16
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import com.google.api.core.ApiFuture;
2121
import com.google.api.core.ApiFutureCallback;
2222
import com.google.api.core.ApiFutures;
23+
import com.google.api.gax.retrying.RetrySettings;
2324
import com.google.cloud.bigquery.storage.v1.AppendRowsResponse;
2425
import com.google.cloud.bigquery.storage.v1.BigQueryWriteClient;
2526
import com.google.cloud.bigquery.storage.v1.CreateWriteStreamRequest;
@@ -37,6 +38,7 @@
3738
import javax.annotation.concurrent.GuardedBy;
3839
import org.json.JSONArray;
3940
import org.json.JSONObject;
41+
import org.threeten.bp.Duration;
4042

4143
public class WriteCommittedStream {
4244

@@ -113,11 +115,25 @@ void initialize(TableName parentTable, BigQueryWriteClient client)
113115
.build();
114116
WriteStream writeStream = client.createWriteStream(createWriteStreamRequest);
115117

118+
// Configure in-stream automatic retry settings.
119+
// Error codes that are immediately retried:
120+
// * ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
121+
// Error codes that are retried with exponential backoff:
122+
// * RESOURCE_EXHAUSTED
123+
RetrySettings retrySettings =
124+
RetrySettings.newBuilder()
125+
.setInitialRetryDelay(Duration.ofMillis(500))
126+
.setRetryDelayMultiplier(1.1)
127+
.setMaxAttempts(5)
128+
.setMaxRetryDelay(Duration.ofMinutes(1))
129+
.build();
130+
116131
// Use the JSON stream writer to send records in JSON format.
117132
// For more information about JsonStreamWriter, see:
118133
// https://googleapis.dev/java/google-cloud-bigquerystorage/latest/com/google/cloud/bigquery/storage/v1/JsonStreamWriter.html
119134
streamWriter =
120135
JsonStreamWriter.newBuilder(writeStream.getName(), writeStream.getTableSchema(), client)
136+
.setRetrySettings(retrySettings)
121137
.build();
122138
}
123139

‎samples/snippets/src/main/java/com/example/bigquerystorage/WritePendingStream.java

+18-2
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,10 @@
1717
package com.example.bigquerystorage;
1818

1919
// [START bigquerystorage_jsonstreamwriter_pending]
20-
2120
import com.google.api.core.ApiFuture;
2221
import com.google.api.core.ApiFutureCallback;
2322
import com.google.api.core.ApiFutures;
23+
import com.google.api.gax.retrying.RetrySettings;
2424
import com.google.cloud.bigquery.storage.v1.AppendRowsResponse;
2525
import com.google.cloud.bigquery.storage.v1.BatchCommitWriteStreamsRequest;
2626
import com.google.cloud.bigquery.storage.v1.BatchCommitWriteStreamsResponse;
@@ -41,6 +41,7 @@
4141
import javax.annotation.concurrent.GuardedBy;
4242
import org.json.JSONArray;
4343
import org.json.JSONObject;
44+
import org.threeten.bp.Duration;
4445

4546
public class WritePendingStream {
4647

@@ -129,6 +130,19 @@ void initialize(TableName parentTable, BigQueryWriteClient client)
129130
// https://googleapis.dev/java/google-cloud-bigquerystorage/latest/com/google/cloud/bigquery/storage/v1/WriteStream.Type.html
130131
WriteStream stream = WriteStream.newBuilder().setType(WriteStream.Type.PENDING).build();
131132

133+
// Configure in-stream automatic retry settings.
134+
// Error codes that are immediately retried:
135+
// * ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
136+
// Error codes that are retried with exponential backoff:
137+
// * RESOURCE_EXHAUSTED
138+
RetrySettings retrySettings =
139+
RetrySettings.newBuilder()
140+
.setInitialRetryDelay(Duration.ofMillis(500))
141+
.setRetryDelayMultiplier(1.1)
142+
.setMaxAttempts(5)
143+
.setMaxRetryDelay(Duration.ofMinutes(1))
144+
.build();
145+
132146
CreateWriteStreamRequest createWriteStreamRequest =
133147
CreateWriteStreamRequest.newBuilder()
134148
.setParent(parentTable.toString())
@@ -140,7 +154,9 @@ void initialize(TableName parentTable, BigQueryWriteClient client)
140154
// For more information about JsonStreamWriter, see:
141155
// https://googleapis.dev/java/google-cloud-bigquerystorage/latest/com/google/cloud/bigquery/storage/v1beta2/JsonStreamWriter.html
142156
streamWriter =
143-
JsonStreamWriter.newBuilder(writeStream.getName(), writeStream.getTableSchema()).build();
157+
JsonStreamWriter.newBuilder(writeStream.getName(), writeStream.getTableSchema())
158+
.setRetrySettings(retrySettings)
159+
.build();
144160
}
145161

146162
public void append(JSONArray data, long offset)

‎samples/snippets/src/main/java/com/example/bigquerystorage/WriteToDefaultStream.java

+19-38
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,11 @@
1717
package com.example.bigquerystorage;
1818

1919
// [START bigquerystorage_jsonstreamwriter_default]
20-
2120
import com.google.api.core.ApiFuture;
2221
import com.google.api.core.ApiFutureCallback;
2322
import com.google.api.core.ApiFutures;
2423
import com.google.api.gax.core.FixedExecutorProvider;
24+
import com.google.api.gax.retrying.RetrySettings;
2525
import com.google.cloud.bigquery.BigQuery;
2626
import com.google.cloud.bigquery.BigQueryOptions;
2727
import com.google.cloud.bigquery.QueryJobConfiguration;
@@ -35,12 +35,9 @@
3535
import com.google.cloud.bigquery.storage.v1.Exceptions.StorageException;
3636
import com.google.cloud.bigquery.storage.v1.JsonStreamWriter;
3737
import com.google.cloud.bigquery.storage.v1.TableName;
38-
import com.google.common.collect.ImmutableList;
3938
import com.google.common.util.concurrent.MoreExecutors;
4039
import com.google.protobuf.ByteString;
4140
import com.google.protobuf.Descriptors.DescriptorValidationException;
42-
import io.grpc.Status;
43-
import io.grpc.Status.Code;
4441
import java.io.IOException;
4542
import java.util.Map;
4643
import java.util.concurrent.Executors;
@@ -49,6 +46,7 @@
4946
import javax.annotation.concurrent.GuardedBy;
5047
import org.json.JSONArray;
5148
import org.json.JSONObject;
49+
import org.threeten.bp.Duration;
5250

5351
public class WriteToDefaultStream {
5452

@@ -97,7 +95,7 @@ public static void writeToDefaultStream(String projectId, String datasetName, St
9795
jsonArr.put(record);
9896
}
9997

100-
writer.append(new AppendContext(jsonArr, 0));
98+
writer.append(new AppendContext(jsonArr));
10199
}
102100

103101
// Final cleanup for the stream during worker teardown.
@@ -130,26 +128,15 @@ private static void verifyExpectedRowCount(TableName parentTable, int expectedRo
130128
private static class AppendContext {
131129

132130
JSONArray data;
133-
int retryCount = 0;
134131

135-
AppendContext(JSONArray data, int retryCount) {
132+
AppendContext(JSONArray data) {
136133
this.data = data;
137-
this.retryCount = retryCount;
138134
}
139135
}
140136

141137
private static class DataWriter {
142138

143-
private static final int MAX_RETRY_COUNT = 3;
144139
private static final int MAX_RECREATE_COUNT = 3;
145-
private static final ImmutableList<Code> RETRIABLE_ERROR_CODES =
146-
ImmutableList.of(
147-
Code.INTERNAL,
148-
Code.ABORTED,
149-
Code.CANCELLED,
150-
Code.FAILED_PRECONDITION,
151-
Code.DEADLINE_EXCEEDED,
152-
Code.UNAVAILABLE);
153140

154141
// Track the number of in-flight requests to wait for all responses before shutting down.
155142
private final Phaser inflightRequestCount = new Phaser(1);
@@ -163,6 +150,19 @@ private static class DataWriter {
163150

164151
public void initialize(TableName parentTable)
165152
throws DescriptorValidationException, IOException, InterruptedException {
153+
// Configure in-stream automatic retry settings.
154+
// Error codes that are immediately retried:
155+
// * ABORTED, UNAVAILABLE, CANCELLED, INTERNAL, DEADLINE_EXCEEDED
156+
// Error codes that are retried with exponential backoff:
157+
// * RESOURCE_EXHAUSTED
158+
RetrySettings retrySettings =
159+
RetrySettings.newBuilder()
160+
.setInitialRetryDelay(Duration.ofMillis(500))
161+
.setRetryDelayMultiplier(1.1)
162+
.setMaxAttempts(5)
163+
.setMaxRetryDelay(Duration.ofMinutes(1))
164+
.build();
165+
166166
// Use the JSON stream writer to send records in JSON format. Specify the table name to write
167167
// to the default stream.
168168
// For more information about JsonStreamWriter, see:
@@ -183,6 +183,7 @@ public void initialize(TableName parentTable)
183183
// column, apply the default value to the missing value field.
184184
.setDefaultMissingValueInterpretation(
185185
AppendRowsRequest.MissingValueInterpretation.DEFAULT_VALUE)
186+
.setRetrySettings(retrySettings)
186187
.build();
187188
}
188189

@@ -244,26 +245,6 @@ public void onSuccess(AppendRowsResponse response) {
244245
}
245246

246247
public void onFailure(Throwable throwable) {
247-
// If the wrapped exception is a StatusRuntimeException, check the state of the operation.
248-
// If the state is INTERNAL, CANCELLED, or ABORTED, you can retry. For more information,
249-
// see: https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html
250-
Status status = Status.fromThrowable(throwable);
251-
if (appendContext.retryCount < MAX_RETRY_COUNT
252-
&& RETRIABLE_ERROR_CODES.contains(status.getCode())) {
253-
appendContext.retryCount++;
254-
try {
255-
// Since default stream appends are not ordered, we can simply retry the appends.
256-
// Retrying with exclusive streams requires more careful consideration.
257-
this.parent.append(appendContext);
258-
// Mark the existing attempt as done since it's being retried.
259-
done();
260-
return;
261-
} catch (Exception e) {
262-
// Fall through to return error.
263-
System.out.format("Failed to retry append: %s\n", e);
264-
}
265-
}
266-
267248
if (throwable instanceof AppendSerializationError) {
268249
AppendSerializationError ase = (AppendSerializationError) throwable;
269250
Map<Integer, String> rowIndexToErrorMessage = ase.getRowIndexToErrorMessage();
@@ -282,7 +263,7 @@ public void onFailure(Throwable throwable) {
282263
// avoid potentially blocking while we are in a callback.
283264
if (dataNew.length() > 0) {
284265
try {
285-
this.parent.append(new AppendContext(dataNew, 0));
266+
this.parent.append(new AppendContext(dataNew));
286267
} catch (DescriptorValidationException e) {
287268
throw new RuntimeException(e);
288269
} catch (IOException e) {

0 commit comments

Comments
 (0)