gemini-code-assist[bot] commented on code in PR #38854:
URL: https://github.com/apache/beam/pull/38854#discussion_r3377564634
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsTableRow.java:
##########
@@ -152,6 +151,7 @@ public StorageApiWritePayload toMessage(
public void updateSchemaFromTable() throws IOException,
InterruptedException {
SCHEMA_CACHE.refreshSchema(
delegate.get().tableReference, datasetService, writeStreamService,
bigQueryOptions);
+ T
Review Comment:
There is a stray 'T' on line 154, which is a syntax error and will cause a
compilation failure. This line should be removed.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java:
##########
@@ -1030,7 +1035,32 @@ public void process(
onSuccess,
appendClientHolder.get(),
tableReference);
+
if (createRetryManagerResult.getSchemaMismatchSeen()) {
+ if (autoUpdateSchemaStrictTimeout != null) {
+ if
(startMismatchTime.plus(initialMismatchRetryTime).isBefore(org.joda.time.Instant.now()))
{
+ // Local retry time has expired!
+ org.joda.time.Instant targetExpirationTime =
startMismatchTime.plus(autoUpdateSchemaStrictTimeout);
+ Iterable<SplittingIterable.Value> mismatchedRows =
+ Iterables.transform(messages,
SplittingIterable.Value::getSchemaMismatchedRowsOnly);
+ bufferMismatchedRows(mismatchedRows, targetExpirationTime)
Review Comment:
There is a missing semicolon at the end of the bufferMismatchedRows call,
which will cause a compilation failure.
```suggestion
bufferMismatchedRows(mismatchedRows, targetExpirationTime);
```
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java:
##########
@@ -849,23 +860,20 @@ public void process(
TableRowToStorageApiProto.descriptorSchemaFromTableSchema(
tableSchema, true, false);
} else {
- // Start off with the base schema. As we get notified of schema
updates, we
- // will update the descriptor.
+ if (autoUpdateSchemaStrictTimeout != null) {
+ messageConverter.updateSchemaFromTable();
Review Comment:
The method updateSchemaFromTable() is called on messageConverter at lines
864 and 1116, but this method is not defined in the
StorageApiDynamicDestinations.MessageConverter interface. This will cause a
compilation failure. To resolve this, please add updateSchemaFromTable() to the
MessageConverter interface in StorageApiDynamicDestinations.java and provide
appropriate implementations in all classes implementing this interface.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java:
##########
@@ -1030,7 +1035,32 @@ public void process(
onSuccess,
appendClientHolder.get(),
tableReference);
+
if (createRetryManagerResult.getSchemaMismatchSeen()) {
+ if (autoUpdateSchemaStrictTimeout != null) {
+ if
(startMismatchTime.plus(initialMismatchRetryTime).isBefore(org.joda.time.Instant.now()))
{
+ // Local retry time has expired!
+ org.joda.time.Instant targetExpirationTime =
startMismatchTime.plus(autoUpdateSchemaStrictTimeout);
+ Iterable<SplittingIterable.Value> mismatchedRows =
+ Iterables.transform(messages,
SplittingIterable.Value::getSchemaMismatchedRowsOnly);
+ bufferMismatchedRows(mismatchedRows, targetExpirationTime)
Review Comment:
The method bufferMismatchedRows is called here, but its definition is
missing from this class and the rest of the PR. Please ensure that this method
is implemented and included in the pull request.
--
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]
