You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/09/19 15:03:20 UTC
[GitHub] [beam] ahmedabu98 opened a new issue, #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
ahmedabu98 opened a new issue, #23291:
URL: https://github.com/apache/beam/issues/23291
### What happened?
When streaming with Storage Write API and using `.withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())`, InvalidArgumentExceptions (eg. schema mismatch) are continuously retried. A schema mismatch is not transient and so the row should not be retried.
### Issue Priority
Priority: 3
### Issue Component
Component: io-java-gcp
--
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: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1325125643
@reuvenlax although this would definitely help in one specific scenario (a required field with default is removed), there are unfortunately other problematic cases where the Avro schema is not in sync with BigQuery (see below for full details, but most notably adding a new optional field is very common for us).
**Note**: in all of the below examples where BigQueryIO returns `INVALID_ARGUMENT`, the errors also end up in the retry-loop discussed earlier and are not immediately visible via `getFailedStorageApiInserts`. Ideally, until we get more flexibility around `SchemaUpdateOption` it would be great if these could be caught immediately and then returned via `getFailedStorageApiInserts`
Scenario | Supported in Avro (Full Transitive) | Supported in BigQueryIO | Comments
-- | -- | -- | --
Add required field with default value | Yes | No | INVALID_ARGUMENT: Input schema has more fields than BigQuery schema
Add optional field with default value | Yes | No | INVALID_ARGUMENT: Input schema has more fields than BigQuery schema
Add default value to existing required field | Yes | No | BigQueryIO runs, but does not sync the new default value
Add default value to existing optional field | Yes | No | BigQueryIO runs, but does not sync the new default value
Remove default value from existing required field | Yes (if field existed continuously since v1) | No | BigQueryIO runs, but does not remove the default value
Remove default value from existing optional field | Yes (if field existed continuously since v1) | No | BigQueryIO runs, but does not remove the default value
Delete optional field | Yes (if default exists) | Yes |
Delete required field | Yes (if default exists) | No | INVALID_ARGUMENT: The required BigQuery field .required_field_with_default is missing in the proto message
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1327993726
Unfortunately a precondition of the sink is that the schema passed into BigQueryIO.write (in your case via getSchema) is (or is compatible with) the schema of the BigQuery table.
In this particular case, the failure doesn't come when we write any specific record. The failure comes when we first open the connection to BigQuery - the connection object takes in the expected schema as a parameter and fails if the schema isn't compatible. Since this is before we start writing records, there is nothing to send to getFailedStorageApiInserts
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1308585619
@ahmedabu98 do you know if this ticket it likely to be addressed anytime soon? This is currently causing us a delay of several hours before failures appear using `getFailedStorageApiInserts`, due to the internal retries.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1320973501
I'm not clear how this would happen. Omitting a required field should fail at the convert-messages stage (and go to getFailedStorageApiInserts) and the BQ write should never even happen. The errorMessage you posted (SchemaDoesntMatchException) is from that stage.
Are you able to provide a simple pipeline that reproduces this?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1322208525
@reuvenlax thanks for taking a look at this. From my experiments, it seems that this can occur when the schema set in BigQueryIO does not match the schema in BigQuery itself. For example, let's assume we have a table with two columns:
1. `required_field` - REQUIRED
2. `optional_field` - NULLABLE
If we set the schema in BigQueryIO to only have the `optional_field` (not `required_field`), then we end up in the retry-loop I mentioned above, because the error is coming back from BigQuery itself rather than being detected up-front by BigQueryIO (which would correctly send it to getFailedStorageApiInserts as expected).
In our use case this can easily happen, because the schemas are set externally in a schema registry (in our real app we use `DynamicDestinations` and `getSchema` makes a dynamic call to our schema registry). Overall, our goal is to update BigQuery to match the new schema as it changes (hence why I've also raised https://github.com/apache/beam/issues/24063).
Rough working example below. Hopefully this makes sense?
```
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryStorageApiInsertError;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.List;
public class BQConnectDemo {
private static final Logger log = LoggerFactory.getLogger(BQConnectDemo.class);
static void runDemo(BQConnectOptions options) {
Pipeline p = Pipeline.create(options);
final List<String> rowText = Arrays.asList(
"To be, or not to be: that is the question: ",
"Whether 'tis nobler in the mind to suffer ",
"The slings and arrows of outrageous fortune, ",
"Or to take arms against a sea of troubles, ");
final PCollection<String> rows = p.apply(Create.of(rowText).withCoder(StringUtf8Coder.of()));
WriteResult writeResult = rows.apply(
"WriteToBigQuery",
BigQueryIO.<String>write()
//[project_id]:[dataset_id].[table_id]
.to("<PROJECT_REDACTED>.<DATASET_REDACTED>.required_fields_demo")
.withSchema(new TableSchema()
.setFields(
List.of(
// Note that schema does not contain required_field
new TableFieldSchema()
.setName("optional_field")
.setType("STRING")
.setMode("NULLABLE")
)
)
)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
.withExtendedErrorInfo()
.withMethod(BigQueryIO.Write.Method.STORAGE_API_AT_LEAST_ONCE)
.withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
.withFormatFunction(new SerializableFunction<String, TableRow>() {
@Override
public TableRow apply(String input) {
// Note that row does not contain required_field
return new TableRow().set("optional_field", "test_value");
}
})
);
PCollection<BigQueryStorageApiInsertError> failedInserts = writeResult.getFailedStorageApiInserts();
failedInserts.apply("Log failures", ParDo.of(new LogErrors()));
p.run();
}
public static void main(String[] args) {
BQConnectOptions options =
PipelineOptionsFactory
.fromArgs(args)
.withValidation()
.as(BQConnectOptions.class);
runDemo(options);
}
static class LogErrors extends DoFn<BigQueryStorageApiInsertError, Void> {
@ProcessElement
public void processElement(ProcessContext context) {
log.error(String.format("Failed to insert element to BigQuery: %s", context.element()));
}
}
}
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1324145656
@reuvenlax interestingly, we see the same behaviour either way, so even when required_field has a default (we had to set one manually, as the auto-created table didn’t have one), we still get the same InvalidArgumentException due to the missing field.
I’m not sure if this is a BigQuery/Storage Write limitation, because in theory it should be possible to insert new data without this field once a default is set?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1311452312
@ahmedabu98 thanks for replying. You're totally correct that `withFailedInsertRetryPolicy()` doesn't apply to STORAGE_WRITE_API and these settings are ignored, however we _are_ still seeing cases where InvalidArgumentExceptions are unnecessarily retried and take several hours to appear via `getFailedStorageApiInserts()`, for example:
1. BigQueryIO receives a TableRow which does not contain a required field in BigQuery (in the below example the required column is called `required_field` for testing purposes). As an aside, it would be fantastic if BigQueryIO could automatically downgrade required columns to nullable, via `SchemaUpdateOptions`, but this is a separate feature request I've raised under https://github.com/apache/beam/issues/24063 and not directly relevant here.
2. The insert to BigQuery fails, returning `io.grpc.StatusRuntimeException: INVALID_ARGUMENT: The required BigQuery field .required_field is missing in the proto message`
3. Unfortunately the error handling you [mentioned](https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java#L448) does not apply in this case, because the error is not an instance of `Exceptions.AppendSerializtionError` (the code would need to check `failedContext.getError() instanceof InvalidArgumentException` for this to work)
4. BigQueryIO skips over the error handling and falls back to logging the failure [here](https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java#L494). Note that this returns `RetryType.RETRY_ALL_OPERATIONS`, meaning that the failure will be retried 1000 times (with backoff) as per the [RetryManager defaults.](https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java#L577)
Hope this helps to clarify? I'm not sure if it makes sense to re-open this issue with an updated description, or create a new issue to capture this?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1323644349
Thanks @reuvenlax, we're using a cached version of our schema registry, so it will only be calling out externally for schemas it hasn't encountered previously.
In terms of how we've seen this error happen in the wild, the workflow is roughly:
1. User creates a new schema in our schema registry. Let's say that initially this contains our two fields above (`required_field` - REQUIRED and `optional_field` - NULLABLE).
2. User starts sending Kafka events (in Avro) using this new schema.
3. BigQueryIO sees the events, then converts from Avro to TableRow using `withFormatFunction`. It pulls the schema from our schema registry via `getSchema` in `DynamicDestinations`, then creates the new table in BigQuery (because it doesn't exist and we are using `BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED`).
4. As new data comes in, BigQueryIO successfully writes the new rows.
5. User makes a change to the schema and submits a new version to the schema registry. In this case, `required_field` was removed, which in Avro terms is valid, so long as there is a default value set.
6. User starts sending Kafka events (in Avro) using the updated schema.
7. BigQueryIO sees the event, then pulls the updated schema via `getSchema`, but then hits the `InvalidArgumentException` issue because BigQuery still expects `required_field` to be there, hence we end up in a retry loop.
In the above case, the ideal outcome for us is that we could set `BigQueryIO.Write.SchemaUpdateOption.ALLOW_FIELD_RELAXATION` and have BigQueryIO automatically downgrade the column to NULLABLE then retry (as per https://github.com/apache/beam/issues/24063), but in the absence of supporting this it would be fantastic if we could get the failures back immediately, because this would allow us to do a full join between the raw Kafka input PCollection and the `PCollection<BigQueryStorageApiInsertError>` from `getFailedStorageApiInserts` (currently this isn't really feasible due to the large time window we'd have to buffer errors). This would then allow us to write the failed inserts back to our Kafka DLQ (as per discussion on https://github.com/apache/beam/issues/24090)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1330485358
@reuvenlax re: `The failure comes when we first open the connection to BigQuery`, from my experiments using the code I posted earlier, this might not always be the case, e.g. if we try to write a TableRow to BigQuery which is missing a required field, it _does_ get caught by the error handling [here](https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java#L437), which means in theory we do have access to the `AppendRowsContext` and the failed `ProtoRows` at this point. Would it be possible to extend the error handling here to not only check for `failedContext.getError() instanceof Exceptions.AppendSerializtionError` but also instances of `InvalidArgumentException`? This would allow us to write the failed rows back out to `getFailedStorageApiInserts` without the 1k retries
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
ahmedabu98 commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1311009258
Hey @carlpayne. I made a mistake creating this bug, as `withFailedInsertRetryPolicy()` is only used with STREAMING_INSERTS, not STORAGE_WRITE_API.
Can you tell me more about your case? What version of Beam are you using? Are you having these problems at HEAD? I saw a recent PR #23556 that improved error handling. The Storage API sink now [removes failed rows before retrying](https://github.com/apache/beam/blob/92cef32915c40d353b51529a87d8f0131b0a3538/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java#L472). FYI these improvements were committed relatively recently.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1322579939
This is an interesting situation. The contract with Storage API is that the descriptor passed in when you open the connection is compatible with the actual BQ table schema, and this descriptor is derived from the schema returned by getSchema. In this case, BigQuery is failing things when we establish the connection (due to the schema mismatch) and before we've sent a single record, which is why things aren't going to the failedInserts collection.
In general, this seems like an unsupported feature. The schema returned by getSchema must be compatible with the actual BQ table schema. If it does not, various things could go wrong in strange ways.
Can you explain how things get out of sync here? Adding new REQUIRED columns to a table is not allowed by BigQuery, so this isn't simply a case where the schema service is returning and old value for the schema.
Another note: if you are calling an external service in getSchema, make sure the value is well cached locally. getSchema is called potentially on every record, so this could cause major performance issues in your pipeline.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
ahmedabu98 commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1311010151
Closing this issue as it's not valid, but feel free to reply here
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1324123206
In this scenario, does the BigQuery schema have a default value set for the required field? https://cloud.google.com/bigquery/docs/default-values
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1324207821
I believe there is currently a bug in Storage Write where they fail the connection even if the required field has a default. This is a bug that I think can be fixed at the BQ level. Would that address your issue?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] closed issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
URL: https://github.com/apache/beam/issues/23291
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
ahmedabu98 commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1311010269
.close-issue
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] carlpayne commented on issue #23291: [Bug]: BQ Storage Write API should treat InvalidArgumentException as a non-transient error.
Posted by GitBox <gi...@apache.org>.
carlpayne commented on issue #23291:
URL: https://github.com/apache/beam/issues/23291#issuecomment-1329373000
Thanks @reuvenlax this does make sense. In that case, it might be a good idea to keep the existing retry behaviour, but from our side we'll probably look to build something that can (asynchronously) compare the schema of incoming messages, then perform the required updates to the BigQuery table. That way, the BigQuery schema will _eventually_ match the record we are trying to write, which should allow the retry to succeed.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org