Auto-create tables when batch insertion fails. #248
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,6 +19,7 @@ | |
|
|
||
|
|
||
| import com.google.cloud.bigquery.BigQuery; | ||
| import com.google.cloud.bigquery.TableId; | ||
| import com.google.cloud.bigquery.BigQueryError; | ||
| import com.google.cloud.bigquery.BigQueryException; | ||
| import com.google.cloud.bigquery.InsertAllRequest; | ||
|
|
@@ -37,16 +38,20 @@ | |
| import java.util.Map; | ||
|
|
||
| /** | ||
| * A {@link BigQueryWriter} capable of updating BigQuery table schemas and creating non-existed tables automatically. | ||
| */ | ||
| public class AdaptiveBigQueryWriter extends BigQueryWriter { | ||
| private static final Logger logger = LoggerFactory.getLogger(AdaptiveBigQueryWriter.class); | ||
|
|
||
| // The maximum number of retries we will attempt to write rows after creating a table or updating a BQ table schema. | ||
| private static final int RETRY_LIMIT = 5; | ||
| // Wait for about 30s between each retry since both creating table and updating schema take up to 2~3 minutes to take effect. | ||
| private static final int RETRY_WAIT_TIME = 30000; | ||
|
|
||
| private final BigQuery bigQuery; | ||
| private final SchemaManager schemaManager; | ||
| private final boolean autoUpdateSchemas; | ||
| private final boolean autoCreateTables; | ||
|
|
||
| /** | ||
| * @param bigQuery Used to send write requests to BigQuery. | ||
|
|
@@ -57,10 +62,14 @@ public class AdaptiveBigQueryWriter extends BigQueryWriter { | |
| public AdaptiveBigQueryWriter(BigQuery bigQuery, | ||
| SchemaManager schemaManager, | ||
| int retry, | ||
| long retryWait, | ||
| boolean autoUpdateSchemas, | ||
| boolean autoCreateTables) { | ||
| super(retry, retryWait); | ||
| this.bigQuery = bigQuery; | ||
| this.schemaManager = schemaManager; | ||
| this.autoUpdateSchemas = autoUpdateSchemas; | ||
| this.autoCreateTables = autoCreateTables; | ||
| } | ||
|
|
||
| private boolean isTableMissingSchema(BigQueryException exception) { | ||
|
|
@@ -69,6 +78,12 @@ private boolean isTableMissingSchema(BigQueryException exception) { | |
| return exception.getReason() != null && exception.getReason().equalsIgnoreCase("invalid"); | ||
| } | ||
|
|
||
| private boolean isTableNotExistedException(BigQueryException exception) { | ||
| // If a table does not exist, it will raise a BigQueryException that the input is notFound | ||
| // Referring to Google Cloud Error Codes Doc: https://cloud.google.com/bigquery/docs/error-messages?hl=en | ||
| return exception.getCode() == 404; | ||
| } | ||
|
|
||
| /** | ||
| * Sends the request to BigQuery, then checks the response to see if any errors have occurred. If | ||
| * any have, and all errors can be blamed upon invalid columns in the rows sent, attempts to | ||
|
|
@@ -86,21 +101,24 @@ public Map<Long, List<BigQueryError>> performWriteRequest( | |
| try { | ||
| request = createInsertAllRequest(tableId, rows); | ||
| writeResponse = bigQuery.insertAll(request); | ||
| // Should only perform one schema update attempt. | ||
| if (writeResponse.hasErrors() | ||
| && onlyContainsInvalidSchemaErrors(writeResponse.getInsertErrors()) && autoUpdateSchemas) { | ||
| attemptSchemaUpdate(tableId, topic); | ||
| } | ||
| } catch (BigQueryException exception) { | ||
| // Should only perform one table creation attempt. | ||
| if (isTableNotExistedException(exception) && autoCreateTables && bigQuery.getTable(tableId.getBaseTableId()) == null) { | ||
| attemptTableCreate(tableId.getBaseTableId(), topic); | ||
| } else if (isTableMissingSchema(exception) && autoUpdateSchemas) { | ||
| attemptSchemaUpdate(tableId, topic); | ||
| } else { | ||
| throw exception; | ||
| } | ||
| } | ||
|
|
||
| // Creating tables or updating table schemas in BigQuery takes up to 2~3 minutes to take affect, | ||
| // so multiple insertion attempts may be necessary. | ||
| int attemptCount = 0; | ||
| while (writeResponse == null || writeResponse.hasErrors()) { | ||
| logger.trace("insertion failed"); | ||
|
|
@@ -117,10 +135,15 @@ && onlyContainsInvalidSchemaErrors(writeResponse.getInsertErrors())) { | |
| return writeResponse.getInsertErrors(); | ||
| } | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment. Updated the doc accordingly~ |
||
| attemptCount++; | ||
| if (attemptCount >= RETRY_LIMIT) { | ||
| throw new BigQueryConnectException( | ||
| "Failed to write rows after BQ schema update within " | ||
| + RETRY_LIMIT + " attempts for: " + tableId.getBaseTableId()); | ||
| } | ||
| try { | ||
| Thread.sleep(RETRY_WAIT_TIME); | ||
|
There was a problem hiding this comment. If retry is caused by schema update, we don't need There was a problem hiding this comment. According to the comment here https://github.com/wepay/kafka-connect-bigquery/blob/master/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/write/row/AdaptiveBigQueryWriter.java#L90-L91, updating schema in bigquery is also an async process which takes up to 2 min. I think we'll want to wait for some time (like 30s) for each retry for updating schema as well. There was a problem hiding this comment. nit: I would suggest add more comment in the function or clean up a little bit. It is not very easy to read to people unfamiliar. There was a problem hiding this comment. Cleaned up the existed comments to my best effort. |
||
| } catch (InterruptedException e) { | ||
| // no-op, we want to keep retrying the insert | ||
| } | ||
| } | ||
| logger.debug("table insertion completed successfully"); | ||
|
|
@@ -136,6 +159,16 @@ private void attemptSchemaUpdate(PartitionedTableId tableId, String topic) { | |
| } | ||
| } | ||
|
|
||
| private void attemptTableCreate(TableId tableId, String topic) { | ||
| try { | ||
| schemaManager.createTable(tableId, topic); | ||
| logger.info("Table {} does not exist, auto-created table for topic {}", tableId, topic); | ||
| } catch (BigQueryException exception) { | ||
| throw new BigQueryConnectException( | ||
| "Failed to create table " + tableId, exception); | ||
| } | ||
| } | ||
|
|
||
| /* | ||
| * Currently, the only way to determine the cause of an insert all failure is by examining the map | ||
| * object returned by the insertErrors() method of an insert all response. The only way to | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@bingqinzhou @mtagle @wicknicks Seems like now there is no way to use this connector without specifying
schema registry.getGcsWriter()is being called at start of the connector and which expects schema registry details. I am using the connector forjsondata which doesn't require schema registry and connector is throwing exception.