Handle HTTP 413 as a batch-size error on the insertAll path

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/exception/BigQueryErrorResponses.java

@@ -41,6 +41,7 @@ public class BigQueryErrorResponses {
   private static final int AUTHENTICATION_ERROR_CODE = 401;
   private static final int FORBIDDEN_CODE = 403;
   private static final int NOT_FOUND_CODE = 404;
+  private static final int CONTENT_TOO_LARGE_CODE = 413;
   private static final int INTERNAL_SERVICE_ERROR_CODE = 500;
   private static final int BAD_GATEWAY_CODE = 502;
   private static final int SERVICE_UNAVAILABLE_CODE = 503;
@@ -146,6 +147,23 @@ public static boolean isRequestTooLargeError(BigQueryException exception) {
         && message(exception.getError()).startsWith("Request payload size exceeds the limit: ");
   }
 
+  /**
+   * Google's HTTP frontend can reject a very large request before it reaches the BigQuery API,
+   * answering with an HTTP 413 and an HTML body instead of the documented JSON error that {@link
+   * #isRequestTooLargeError(BigQueryException)} matches. Which of the two responses comes back is
+   * not deterministic: an identical oversized request can be answered with the 400 JSON error on
+   * one attempt and the 413 on the next, so both have to be treated as meaning the payload was too
+   * large.
+   *
+   * <p>A 413 does not always mean the request was too large: per <a
+   * href="https://docs.cloud.google.com/docs/quotas/troubleshoot">Google's quota troubleshooting
+   * guide</a>, the legacy streaming API also returns one when throughput exceeds 300MB/s. This is a
+   * separate edge case that this project does not yet handle.
+   */
+  public static boolean isContentTooLargeError(BigQueryException exception) {
+    return CONTENT_TOO_LARGE_CODE == exception.getCode();
+  }
+
   public static boolean isTooManyRowsError(BigQueryException exception) {
     return BAD_REQUEST_CODE == exception.getCode()
         && INVALID_REASON.equalsIgnoreCase(exception.getReason())

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/write/batch/TableWriter.java

@@ -119,9 +119,13 @@ private static boolean isBatchSizeError(BigQueryException exception) {
      * documented but is referenced slightly under "Error codes" here:
      * https://cloud.google.com/bigquery/quota-policy
      * (by decreasing the batch size we can eventually expect to end up with a request under 10MB)
+     *
+     * A 413 is the same condition reported by Google's HTTP frontend rather than by BigQuery
+     * itself, which happens intermittently for payloads well above the limit.
      */
     return BigQueryErrorResponses.isUnspecifiedBadRequestError(exception)
         || BigQueryErrorResponses.isRequestTooLargeError(exception)
+        || BigQueryErrorResponses.isContentTooLargeError(exception)
         || BigQueryErrorResponses.isTooManyRowsError(exception);
   }
 

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/exception/BigQueryErrorResponsesTest.java

@@ -97,6 +97,21 @@ public void testIsTableUnavailableError() {
     assertFalse(BigQueryErrorResponses.isTableUnavailableError(error));
   }
 
+  @Test
+  public void testIsContentTooLargeError() {
+    // Google's HTTP frontend answers with an HTML body, so the client populates neither a reason
+    // nor a structured error - only the status code
+    assertTrue(
+        BigQueryErrorResponses.isContentTooLargeError(
+            new BigQueryException(413, "413 Request Entity Too Large")));
+
+    // the 400 form of the same condition is matched by isRequestTooLargeError instead
+    String message = "Request payload size exceeds the limit: 10485760 bytes.";
+    assertFalse(
+        BigQueryErrorResponses.isContentTooLargeError(
+            new BigQueryException(400, message, new BigQueryError("badRequest", null, message))));
+  }
+
   @Test
   public void testIsAuthenticationError() {
     BigQueryException error = new BigQueryException(0, "......401.....Unauthorized error.....");