prodriguezdefino commented on code in PR #17550:
URL: https://github.com/apache/beam/pull/17550#discussion_r868563750
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -263,9 +295,9 @@ void invalidateWriteStream() {
// thread has already invalidated
// and recreated the stream).
@Nullable
- StreamAppendClient cachedAppendClient =
APPEND_CLIENTS.getIfPresent(streamName);
- if (cachedAppendClient != null
- && System.identityHashCode(cachedAppendClient)
+ List<StreamAppendClient> cachedAppendClients =
APPEND_CLIENTS.getIfPresent(streamName);
+ if (cachedAppendClients != null
+ &&
System.identityHashCode(cachedAppendClients.get(clientNumber))
== System.identityHashCode(streamAppendClient)) {
Review Comment:
made the changes to only invalidate the writer in use by the bundle, not all
of them.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -357,28 +389,33 @@ void flush(RetryManager<AppendRowsResponse,
Context<AppendRowsResponse>> retryMa
private transient @Nullable DatasetService datasetService;
private int numPendingRecords = 0;
private int numPendingRecordBytes = 0;
- private static final int FLUSH_THRESHOLD_RECORDS = 150000;
private final int flushThresholdBytes;
+ private final int flushThresholdCount;
private final StorageApiDynamicDestinations<ElementT, DestinationT>
dynamicDestinations;
private final BigQueryServices bqServices;
private final boolean useDefaultStream;
+ // default append client count to 1
+ private Integer streamAppendClientCount = 1;
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
