Clean up comments and add missed doc.

Author: Bingqin Zhou

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

@@ -32,7 +32,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Batch Table Writer that uploads records to GCS as a blob
@@ -57,6 +56,7 @@ public class GCSBatchTableWriter implements Runnable {
    * @param bucketName the name of the GCS bucket where the blob should be uploaded
    * @param baseBlobName the base name of the blob in which the serialized rows should be uploaded.
    *                     The full name is [baseBlobName]_[writerId]_
+   * @param topic Kafka record topic
    */
   private GCSBatchTableWriter(List<RowToInsert> rows,
                               GCSToBQWriter writer,
@@ -105,6 +105,7 @@ public static class Builder implements TableWriterBuilder {
      * @param tableId The bigquery table to be written to.
      * @param gcsBucketName The GCS bucket to write to.
      * @param gcsBlobName The name of the GCS blob to write.
+     * @param topic Kafka record topic
      * @param recordConverter the {@link RecordConverter} to use.
      */
     public Builder(GCSToBQWriter writer,

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/write/row/AdaptiveBigQueryWriter.java

@@ -43,9 +43,9 @@
 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 updating a BQ table schema.
-  private static final int AFTER_UPDATE_RETRY_LIMIT = 5;
-  // Wait for about 30s between each retry since both creating table and updating schema take up to 2 minutes to take effect.
+  // 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;
@@ -101,8 +101,7 @@ public Map<Long, List<BigQueryError>> performWriteRequest(
     try {
       request = createInsertAllRequest(tableId, rows);
       writeResponse = bigQuery.insertAll(request);
-      // Should only perform one schema update attempt; may have to continue insert attempts due to
-      // BigQuery schema updates taking up to two minutes to take effect
+      // Should only perform one schema update attempt.
       if (writeResponse.hasErrors()
               && onlyContainsInvalidSchemaErrors(writeResponse.getInsertErrors()) && autoUpdateSchemas) {
         attemptSchemaUpdate(tableId, topic);
@@ -118,7 +117,8 @@ && onlyContainsInvalidSchemaErrors(writeResponse.getInsertErrors()) && autoUpdat
       }
     }
 
-    // Schema update or table creation might be delayed, so multiple insertion attempts may be necessary
+    // 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");
@@ -135,10 +135,10 @@ && onlyContainsInvalidSchemaErrors(writeResponse.getInsertErrors()) && autoUpdat
         return writeResponse.getInsertErrors();
       }
       attemptCount++;
-      if (attemptCount >= AFTER_UPDATE_RETRY_LIMIT) {
+      if (attemptCount >= RETRY_LIMIT) {
         throw new BigQueryConnectException(
             "Failed to write rows after BQ schema update within "
-                + AFTER_UPDATE_RETRY_LIMIT + " attempts for: " + tableId.getBaseTableId());
+                + RETRY_LIMIT + " attempts for: " + tableId.getBaseTableId());
       }
       try {
         Thread.sleep(RETRY_WAIT_TIME);