[Chakravarthy|dV2_BQ_API] - Enables config for setting cleint connect…

… and read timeouts for BQ client (#43)

Google v2 streaming backend allows a increased bandwidth than the previous versions upto 1 GB/Sec at project level. Moving to v2 has produced intermittent
connection failures on timeouts. Based on the recommendation from google, this feature provides a configurable setting for read timeout and connection timeout

src/main/java/com/gojek/beast/config/BQConfig.java

@@ -25,4 +25,12 @@ public interface BQConfig extends Config {
     @DefaultValue("true")
     @Key("ENABLE_BQ_ROW_INSERTID")
     Boolean isBQRowInsertIdEnabled();
+
+    @DefaultValue("-1")
+    @Key("BQ_CLIENT_READ_TIMEOUT")
+    String getBqClientReadTimeout();
+
+    @DefaultValue("-1")
+    @Key("BQ_CLIENT_CONNECT_TIMEOUT")
+    String getBqClientConnectTimeout();
 }

src/main/java/com/gojek/beast/factory/BeastFactory.java

@@ -41,6 +41,7 @@
 import com.gojek.beast.worker.WorkerState;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.auth.oauth2.ServiceAccountCredentials;
+import com.google.cloud.TransportOptions;
 import com.google.cloud.bigquery.BigQuery;
 import com.google.cloud.bigquery.BigQueryOptions;
 import com.google.cloud.bigquery.TableId;
@@ -94,6 +95,7 @@ public List<Worker> createBqWorkers() {
         Integer bqWorkerPoolSize = appConfig.getBqWorkerPoolSize();
         List<Worker> threads = new ArrayList<>(bqWorkerPoolSize);
         Acknowledger acknowledger = createAcknowledger();
+        log.info("BQ Row InsertId is: {}", (bqConfig.isBQRowInsertIdEnabled()) ? "Enabled" : "Disabled");
         for (int i = 0; i < bqWorkerPoolSize; i++) {
             Worker bqQueueWorker = new BqQueueWorker("bq-worker-" + i, createBigQuerySink(), new QueueConfig(appConfig.getBqWorkerPollTimeoutMs()), acknowledger, readQueue, workerState);
             threads.add(bqQueueWorker);
@@ -125,7 +127,12 @@ public BQErrorHandler createOOBErrorHandler() {
     }
 
     private BigQuery getBigQueryInstance() {
+        final TransportOptions transportOptions = BigQueryOptions.getDefaultHttpTransportOptions().toBuilder()
+                .setConnectTimeout(Integer.parseInt(bqConfig.getBqClientConnectTimeout()))
+                .setReadTimeout(Integer.parseInt(bqConfig.getBqClientReadTimeout()))
+                .build();
         return BigQueryOptions.newBuilder()
+                .setTransportOptions(transportOptions)
                 .setCredentials(getGoogleCredentials())
                 .setProjectId(bqConfig.getGCPProject())
                 .build().getService();

src/main/java/com/gojek/beast/models/FailureStatus.java

@@ -28,7 +28,7 @@ public Optional<Exception> getException() {
     @Override
     public String toString() {
         return "FailureStatus{"
-                + "cause=" + cause.getMessage()
-                + ", message='" + message + '\'' + '}';
+                + "cause=" + cause.getCause()
+                + ", message='" + ((message != null) ? message : cause.getMessage()) + '\'' + '}';
     }
 }

src/main/java/com/gojek/beast/sink/executor/RetryExecutor.java

@@ -50,6 +50,7 @@ public Executor execute() {
         try {
             status = sink.push(records);
         } catch (Exception e) {
+            statsClient.increment("retrysink.exec.failure.count," + statsClient.getBqTags());
             status = new FailureStatus(e);
         }
         ifFailure();