Fix gcs load retry logic #170
Conversation
| } catch (ConnectException ex) { | ||
| throw new ConnectException("Failed to write rows to GCS", ex); | ||
| } catch (InterruptedException ex) { | ||
| throw new ConnectException("Thread interrupted while batch writing to GCS.", ex); |
There was a problem hiding this comment.
Is it easy at this point to indicate the name of the connector in the log? Or some identifying mark so we can know which connector is generating the error.
There was a problem hiding this comment.
hm, I'll look into it.
There was a problem hiding this comment.
@mtagle you might look to see if Kafka connect now includes connector info in the log context (MDC). If so, SREs just need to include a new pre-defined variable.
Yup. https://issues.apache.org/jira/browse/KAFKA-3816 .. I recommend just using this once we get to 2.3, since it looks like this is when it was added.
There was a problem hiding this comment.
so I can't give the connector name but I can give the name of the table it's attempting to get data into. That is hopefully helpful?
| logger.info("Batch loaded {} rows", rows.size()); | ||
| } | ||
| else { | ||
| throw new ConnectException(String.format("Failed to load %d rows into GCS within %d attempts.", rows.size(), retries)); |
There was a problem hiding this comment.
Similarly here and in line 128 (logger.info("Batch loaded...). It would be nice to be able to identify the connector that is generating the error.
| } catch (ConnectException ex) { | ||
| throw new ConnectException("Failed to write rows to GCS", ex); | ||
| } catch (InterruptedException ex) { | ||
| throw new ConnectException("Thread interrupted while batch writing to GCS.", ex); |
There was a problem hiding this comment.
@mtagle you might look to see if Kafka connect now includes connector info in the log context (MDC). If so, SREs just need to include a new pre-defined variable.
Yup. https://issues.apache.org/jira/browse/KAFKA-3816 .. I recommend just using this once we get to 2.3, since it looks like this is when it was added.
| logger.warn("Exceptions occurred for table {}, attempting retry", tableId.getTable()); | ||
| } | ||
| retryCount++; | ||
| } while (exceptionsOccurred && (retryCount < retries)); | ||
| } while (!success && (retryCount < retries)); |
There was a problem hiding this comment.
Possible to move this to a normal while() rather than do/while as a best-practices thing?
| if (success) { | ||
| logger.info("Batch loaded {} rows", rows.size()); | ||
| } | ||
| else { |
There was a problem hiding this comment.
Nit: Not sure , but :
}
else {
Seems not to style.
| logger.info("Batch loaded {} rows", rows.size()); | ||
| } | ||
| else { | ||
| throw new ConnectException(String.format("Failed to load %d rows into GCS within %d attempts.", rows.size(), retries)); |
…etries are actually retries and not total tries. Address other comments.
|
LGTM! |
wrote a bunch of tests for GCSToBQWriter
fixed retry logic for GCSToBQWriter:
it's possible that this will fix the issues we've been seeing with dropped rows when going through GCS loading.