Merge pull request #1525 from cloudsufi/patch/PLUGIN-1866

Pushpender Saini · web-flow · commit 64891be0a3f4 · 2025-03-12T15:38:55.000+05:30
[PLUGIN-1866] Add Http readTimeout BQ Source/Sink
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/action/BigQueryExecute.java b/src/main/java/io/cdap/plugin/gcp/bigquery/action/BigQueryExecute.java
@@ -45,7 +45,6 @@
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.cdap.api.annotation.Plugin;
 import io.cdap.cdap.api.exception.ErrorCategory;
-import io.cdap.cdap.api.exception.ErrorCodeType;
 import io.cdap.cdap.api.exception.ErrorType;
 import io.cdap.cdap.api.exception.ErrorUtils;
 import io.cdap.cdap.etl.api.FailureCollector;
@@ -364,7 +363,6 @@ public static final class Config extends AbstractBigQueryActionConfig {
     public static final int DEFAULT_MAX_RETRY_COUNT = 5;
     // Sn = a * (1 - r^n) / (r - 1)
     public static final long DEFULT_MAX_RETRY_DURATION_SECONDS = 63L;
-    public static final int DEFAULT_READ_TIMEOUT = 120;
     public static final Set<String> VALID_WRITE_PREFERENCES = Arrays.stream(JobInfo.WriteDisposition.values())
         .map(Enum::name).collect(Collectors.toSet());
 
@@ -581,7 +579,7 @@ public int getMaxRetryCount() {
     }
 
     public int getReadTimeout() {
-      return readTimeout == null ? DEFAULT_READ_TIMEOUT : readTimeout;
+      return readTimeout == null ? GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS : readTimeout;
     }
 
     @Override
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sink/AbstractBigQuerySink.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sink/AbstractBigQuerySink.java
@@ -103,7 +103,7 @@ public final void prepareRun(BatchSinkContext context) throws Exception {
     DatasetId datasetId = DatasetId.of(config.getDatasetProject(), config.getDataset());
     Dataset dataset;
     try {
-      bigQuery = GCPUtils.getBigQuery(project, credentials, null);
+      bigQuery = GCPUtils.getBigQuery(project, credentials, config.getReadTimeout());
       dataset = bigQuery.getDataset(datasetId);
     } catch (Exception e) {
       throw GCPErrorDetailsProviderUtil.getHttpResponseExceptionDetailsFromChain(e,
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sink/AbstractBigQuerySinkConfig.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sink/AbstractBigQuerySinkConfig.java
@@ -33,6 +33,7 @@
 import io.cdap.plugin.gcp.bigquery.util.BigQueryUtil;
 import io.cdap.plugin.gcp.common.CmekUtils;
 
+import io.cdap.plugin.gcp.common.GCPUtils;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
@@ -54,6 +55,7 @@ public abstract class AbstractBigQuerySinkConfig extends BigQueryBaseConfig {
   protected static final String NAME_UPDATE_SCHEMA = "allowSchemaRelaxation";
   private static final String SCHEME = "gs://";
   protected static final String NAME_JSON_STRING_FIELDS = "jsonStringFields";
+  private static final String NAME_READ_TIMEOUT = "readTimeout";
 
   @Name(Constants.Reference.REFERENCE_NAME)
   @Nullable
@@ -100,6 +102,16 @@ public abstract class AbstractBigQuerySinkConfig extends BigQueryBaseConfig {
           "The schema of these fields should be of type STRING.")
   protected String jsonStringFields;
 
+  @Name(NAME_READ_TIMEOUT)
+  @Nullable
+  @Macro
+  @Description("Timeout in seconds to read data from an established HTTP connection (Default value is 120).")
+  private Integer readTimeout;
+
+  public int getReadTimeout() {
+    return readTimeout == null ? GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS : readTimeout;
+  }
+
   public AbstractBigQuerySinkConfig(BigQueryConnectorConfig connection, String dataset, String cmekKey, String bucket) {
     super(connection, dataset, cmekKey, bucket);
   }
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQueryMultiSink.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQueryMultiSink.java
@@ -128,7 +128,7 @@ protected void configureOutputSchemas(BatchSinkContext context,
 
         Table table = BigQueryUtil.getBigQueryTable(
           config.getDatasetProject(), config.getDataset(), tableName, config.getServiceAccount(),
-          config.isServiceAccountFilePath(), collector, null);
+          config.isServiceAccountFilePath(), collector, config.getReadTimeout());
 
         Schema tableSchema = configuredSchema;
         if (table != null) {
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQueryMultiSinkConfig.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQueryMultiSinkConfig.java
@@ -20,6 +20,7 @@
 import io.cdap.cdap.api.annotation.Macro;
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.plugin.gcp.bigquery.connector.BigQueryConnectorConfig;
+import io.cdap.plugin.gcp.common.GCPUtils;
 
 import javax.annotation.Nullable;
 
@@ -30,6 +31,7 @@ public class BigQueryMultiSinkConfig extends AbstractBigQuerySinkConfig {
 
   private static final String SPLIT_FIELD_DEFAULT = "tablename";
   private static final String NAME_ALLOW_FLEXIBLE_SCHEMA = "allowFlexibleSchema";
+  private static final String NAME_READ_TIMEOUT = "readTimeout";
 
   @Macro
   @Nullable
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQuerySink.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQuerySink.java
@@ -138,7 +138,7 @@ protected void prepareRunInternal(BatchSinkContext context, BigQuery bigQuery, S
     configureBigQuerySink();
     Table table = BigQueryUtil.getBigQueryTable(config.getDatasetProject(), config.getDataset(), config.getTable(),
             config.getServiceAccount(), config.isServiceAccountFilePath(),
-            collector, null);
+            collector, config.getReadTimeout());
     initOutput(context, bigQuery, config.getReferenceName(),
                BigQueryUtil.getFQN(config.getDatasetProject(), config.getDataset(), config.getTable()),
                config.getTable(), outputSchema, bucket, collector, null, table);
@@ -343,10 +343,8 @@ private void configureBigQuerySink() {
    */
   private void configureTable(Schema schema) {
     AbstractBigQuerySinkConfig config = getConfig();
-    Table table = BigQueryUtil.getBigQueryTable(config.getDatasetProject(), config.getDataset(),
-                                                config.getTable(),
-                                                config.getServiceAccount(),
-                                                config.isServiceAccountFilePath(), null, null);
+    Table table = BigQueryUtil.getBigQueryTable(config.getDatasetProject(), config.getDataset(), config.getTable(),
+        config.getServiceAccount(), config.isServiceAccountFilePath(), null, config.getReadTimeout());
     baseConfiguration.setBoolean(BigQueryConstants.CONFIG_DESTINATION_TABLE_EXISTS, table != null);
     List<String> tableFieldsNames = null;
     if (table != null) {
@@ -372,7 +370,7 @@ private void validateConfiguredSchema(Schema schema, FailureCollector collector)
     String tableName = config.getTable();
     Table table = BigQueryUtil.getBigQueryTable(config.getDatasetProject(), config.getDataset(), tableName,
                                                 config.getServiceAccount(), config.isServiceAccountFilePath(),
-                                                collector, null);
+                                                collector, config.getReadTimeout());
 
     if (table != null && !config.containsMacro(AbstractBigQuerySinkConfig.NAME_UPDATE_SCHEMA)) {
       // if table already exists, validate schema against underlying bigquery table
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQuerySinkConfig.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQuerySinkConfig.java
@@ -417,7 +417,7 @@ private void validatePartitionProperties(@Nullable Schema schema, FailureCollect
     }
 
     Table table = BigQueryUtil.getBigQueryTable(project, dataset, tableName, serviceAccount,
-                                                isServiceAccountFilePath(), collector, null);
+        isServiceAccountFilePath(), collector, getReadTimeout());
     if (table != null) {
       StandardTableDefinition tableDefinition = table.getDefinition();
       TimePartitioning timePartitioning = tableDefinition.getTimePartitioning();
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQuerySource.java b/src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQuerySource.java
@@ -260,6 +260,7 @@ private void configureBigQuerySource() {
     if (config.getViewMaterializationDataset() != null) {
       configuration.set(BigQueryConstants.CONFIG_VIEW_MATERIALIZATION_DATASET, config.getViewMaterializationDataset());
     }
+    configuration.set(BigQueryConstants.CONFIG_BQ_HTTP_READ_TIMEOUT, String.valueOf(config.getReadTimeout()));
   }
 
   public Schema getSchema(FailureCollector collector) {
@@ -308,7 +309,7 @@ private com.google.cloud.bigquery.Schema getBQSchema(FailureCollector collector)
     String project = config.getDatasetProject();
 
     Table table = BigQueryUtil.getBigQueryTable(project, dataset, tableName, serviceAccount,
-                                                config.isServiceAccountFilePath(), collector, null);
+        config.isServiceAccountFilePath(), collector, config.getReadTimeout());
     if (table == null) {
       // Table does not exist
       collector.addFailure(String.format("BigQuery table '%s:%s.%s' does not exist.", project, dataset, tableName),
@@ -341,7 +342,7 @@ private void validatePartitionProperties(FailureCollector collector) {
     String dataset = config.getDataset();
     String tableName = config.getTable();
     Table sourceTable = BigQueryUtil.getBigQueryTable(project, dataset, tableName, config.getServiceAccount(),
-                                                      config.isServiceAccountFilePath(), collector, null);
+        config.isServiceAccountFilePath(), collector, config.getReadTimeout());
     if (sourceTable == null) {
       return;
     }
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQuerySourceConfig.java b/src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQuerySourceConfig.java
@@ -58,6 +58,7 @@ public final class BigQuerySourceConfig extends BigQueryBaseConfig {
   private static final String VALID_DATE_FORMAT = "yyyy-MM-dd";
   private static final String SCHEME = "gs://";
   private static final String WHERE = "WHERE";
+  private static final String NAME_READ_TIMEOUT = "readTimeout";
   public static final Set<Schema.Type> SUPPORTED_TYPES =
     ImmutableSet.of(Schema.Type.LONG, Schema.Type.STRING, Schema.Type.DOUBLE, Schema.Type.BOOLEAN, Schema.Type.BYTES,
                     Schema.Type.ARRAY, Schema.Type.RECORD);
@@ -131,6 +132,12 @@ public final class BigQuerySourceConfig extends BigQueryBaseConfig {
     + "Defaults to the same dataset in which the table is located.")
   private String viewMaterializationDataset;
 
+  @Name(NAME_READ_TIMEOUT)
+  @Nullable
+  @Macro
+  @Description("Timeout in seconds to read data from an established HTTP connection (Default value is 120).")
+  private Integer readTimeout;
+
   public String getTable() {
     return table;
   }
@@ -142,6 +149,10 @@ public String getDatasetProject() {
     return connection.getDatasetProject();
   }
 
+  public int getReadTimeout() {
+    return readTimeout == null ? GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS : readTimeout;
+  }
+
   public void validate(FailureCollector collector) {
     validate(collector, Collections.emptyMap());
   }
@@ -238,7 +249,7 @@ private void validatePartitionDate(FailureCollector collector, String partitionD
   public Type getSourceTableType() {
     Table sourceTable =
       BigQueryUtil.getBigQueryTable(getDatasetProject(), getDataset(), table, getServiceAccount(),
-                                    isServiceAccountFilePath(), null, null);
+                                    isServiceAccountFilePath(), null, getReadTimeout());
     return sourceTable != null ? sourceTable.getDefinition().getType() : null;
   }
 
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/source/PartitionedBigQueryInputFormat.java b/src/main/java/io/cdap/plugin/gcp/bigquery/source/PartitionedBigQueryInputFormat.java
@@ -132,9 +132,11 @@ private void processQuery(JobContext context) throws IOException, InterruptedExc
     String partitionFromDate = configuration.get(BigQueryConstants.CONFIG_PARTITION_FROM_DATE, null);
     String partitionToDate = configuration.get(BigQueryConstants.CONFIG_PARTITION_TO_DATE, null);
     String filter = configuration.get(BigQueryConstants.CONFIG_FILTER, null);
+    Integer readTimeout = configuration.getInt(BigQueryConstants.CONFIG_BQ_HTTP_READ_TIMEOUT,
+        GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
 
     com.google.cloud.bigquery.Table bigQueryTable = BigQueryUtil.getBigQueryTable(
-      datasetProjectId, datasetId, tableName, serviceAccount, isServiceAccountFilePath, null, null);
+      datasetProjectId, datasetId, tableName, serviceAccount, isServiceAccountFilePath, null, readTimeout);
     Type type = Objects.requireNonNull(bigQueryTable).getDefinition().getType();
 
     String query;
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryConstants.java b/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryConstants.java
@@ -49,4 +49,5 @@ public interface BigQueryConstants {
   String CDAP_BQ_SINK_OUTPUT_SCHEMA = "cdap.bq.sink.output.schema";
   String BQ_FQN_PREFIX = "bigquery";
   String CONFIG_JOB_LABEL_KEY_VALUE = "cdap.bq.sink.job.label.key.value";
+  String CONFIG_BQ_HTTP_READ_TIMEOUT = "cdap.bq.job.http.read.timeout";
 }
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryUtil.java b/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryUtil.java
@@ -603,7 +603,7 @@ public static Table getBigQueryTable(String projectId, String dataset, String ta
     }
     BigQuery bigQuery = GCPUtils.getBigQuery(projectId, credentials, readTimeoutSeconds);
 
-    Table table = null;
+    Table table;
     try {
       table = bigQuery.getTable(tableId);
     } catch (BigQueryException e) {
diff --git a/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java b/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java
@@ -35,9 +35,12 @@
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 import com.google.gson.reflect.TypeToken;
+import io.cdap.plugin.gcp.bigquery.util.BigQueryUtil;
 import io.cdap.plugin.gcp.gcs.GCSPath;
 import io.cdap.plugin.gcp.gcs.ServiceAccountAccessTokenProvider;
 import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.threeten.bp.Duration;
 
 import java.io.ByteArrayInputStream;
@@ -62,6 +65,8 @@
  * GCP utility class to get service account credentials
  */
 public class GCPUtils {
+
+  private static final Logger LOG = LoggerFactory.getLogger(GCPUtils.class);
   public static final String FS_GS_PROJECT_ID = "fs.gs.project.id";
   private static final Gson GSON = new Gson();
   private static final Type SCOPES_TYPE = new TypeToken<List<String>>() { }.getType();
@@ -82,6 +87,7 @@ public class GCPUtils {
   public static final String GCS_SUPPORTED_DOC_URL = "https://cloud.google.com/storage/docs/json_api/v1/status-codes";
   public static final String BQ_SUPPORTED_DOC_URL = "https://cloud.google.com/bigquery/docs/error-messages";
   public static final String PUBSUB_SUPPORTED_DOC_URL = "https://cloud.google.com/pubsub/docs/reference/error-codes";
+  public static final int BQ_DEFAULT_READ_TIMEOUT_SECONDS = 120;
 
   /**
    * Load a service account from the local file system.
@@ -259,6 +265,7 @@ public static BigQuery getBigQuery(String project, @Nullable Credentials credent
     }
 
     if (readTimeout != null) {
+      LOG.debug("Setting read timeout to {} seconds.", readTimeout);
       bigqueryBuilder.setTransportOptions(HttpTransportOptions.newBuilder()
           .setReadTimeout(readTimeout * MILLISECONDS_MULTIPLIER).build());
     }
diff --git a/src/test/java/io/cdap/plugin/gcp/bigquery/action/BigQueryExecuteTest.java b/src/test/java/io/cdap/plugin/gcp/bigquery/action/BigQueryExecuteTest.java
@@ -31,6 +31,7 @@
 import io.cdap.cdap.etl.api.action.ActionContext;
 
 import io.cdap.cdap.etl.mock.validation.MockFailureCollector;
+import io.cdap.plugin.gcp.common.GCPUtils;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -134,7 +135,7 @@ public void testValidateRetryConfigurationWithDefaultValues() {
             BigQueryExecute.Config.DEFULT_MAX_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFAULT_MAX_RETRY_COUNT,
             BigQueryExecute.Config.DEFAULT_RETRY_MULTIPLIER,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(0, failureCollector.getValidationFailures().size());
   }
 
@@ -144,7 +145,7 @@ public void testValidateRetryConfigurationWithInvalidInitialRetryDuration() {
             BigQueryExecute.Config.DEFULT_MAX_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFAULT_MAX_RETRY_COUNT,
             BigQueryExecute.Config.DEFAULT_RETRY_MULTIPLIER,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(1, failureCollector.getValidationFailures().size());
     Assert.assertEquals("Initial retry duration must be greater than 0.",
             failureCollector.getValidationFailures().get(0).getMessage());
@@ -156,7 +157,7 @@ public void testValidateRetryConfigurationWithInvalidMaxRetryDuration() {
             BigQueryExecute.Config.DEFAULT_INITIAL_RETRY_DURATION_SECONDS, -1L,
             BigQueryExecute.Config.DEFAULT_MAX_RETRY_COUNT,
             BigQueryExecute.Config.DEFAULT_RETRY_MULTIPLIER,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(2, failureCollector.getValidationFailures().size());
     Assert.assertEquals("Max retry duration must be greater than 0.",
             failureCollector.getValidationFailures().get(0).getMessage());
@@ -170,7 +171,7 @@ public void testValidateRetryConfigurationWithInvalidRetryMultiplier() {
             BigQueryExecute.Config.DEFAULT_INITIAL_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFULT_MAX_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFAULT_MAX_RETRY_COUNT, -1.0,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(1, failureCollector.getValidationFailures().size());
     Assert.assertEquals("Retry multiplier must be strictly greater than 1.",
             failureCollector.getValidationFailures().get(0).getMessage());
@@ -182,7 +183,7 @@ public void testValidateRetryConfigurationWithInvalidRetryMultiplierAndMaxRetryC
             BigQueryExecute.Config.DEFAULT_INITIAL_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFULT_MAX_RETRY_DURATION_SECONDS, -1,
             BigQueryExecute.Config.DEFAULT_RETRY_MULTIPLIER,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(1, failureCollector.getValidationFailures().size());
     Assert.assertEquals("Max retry count must be greater than 0.",
             failureCollector.getValidationFailures().get(0).getMessage());
@@ -194,7 +195,7 @@ public void testValidateRetryConfigurationWithMultiplierOne() {
             BigQueryExecute.Config.DEFAULT_INITIAL_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFULT_MAX_RETRY_DURATION_SECONDS,
             BigQueryExecute.Config.DEFAULT_MAX_RETRY_COUNT, 1.0,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(1, failureCollector.getValidationFailures().size());
     Assert.assertEquals("Retry multiplier must be strictly greater than 1.",
             failureCollector.getValidationFailures().get(0).getMessage());
@@ -205,7 +206,7 @@ public void testValidateRetryConfigurationWithMaxRetryLessThanInitialRetry() {
     config.validateRetryConfiguration(failureCollector, 10L, 5L,
             BigQueryExecute.Config.DEFAULT_MAX_RETRY_COUNT,
             BigQueryExecute.Config.DEFAULT_RETRY_MULTIPLIER,
-            BigQueryExecute.Config.DEFAULT_READ_TIMEOUT);
+            GCPUtils.BQ_DEFAULT_READ_TIMEOUT_SECONDS);
     Assert.assertEquals(1, failureCollector.getValidationFailures().size());
     Assert.assertEquals("Max retry duration must be greater than initial retry duration.",
             failureCollector.getValidationFailures().get(0).getMessage());
diff --git a/src/test/java/io/cdap/plugin/gcp/bigquery/source/PartitionedBigQueryInputFormatTest.java b/src/test/java/io/cdap/plugin/gcp/bigquery/source/PartitionedBigQueryInputFormatTest.java
@@ -64,8 +64,7 @@ public void testGenerateQuery() {
     StandardTableDefinition tableDefinition = PowerMockito.mock(StandardTableDefinition.class);
     PowerMockito.when(BigQueryUtil.getBigQueryTable(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(),
         ArgumentMatchers.anyString(), ArgumentMatchers.any(), ArgumentMatchers.anyBoolean(), ArgumentMatchers.any(),
-        ArgumentMatchers.any()))
-      .thenReturn(t);
+        ArgumentMatchers.any())).thenReturn(t);
     PowerMockito.when(t.getDefinition()).thenReturn(tableDefinition);
     String generatedQuery = partitionedBigQueryInputFormat.generateQuery(null, null, filter, datasetProject,
                                                                          datasetProject, dataset, table, null, true);
diff --git a/widgets/BigQueryMultiTable-batchsink.json b/widgets/BigQueryMultiTable-batchsink.json
@@ -205,6 +205,15 @@
               }
             ]
           }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Read Timeout",
+          "name": "readTimeout",
+          "widget-attributes": {
+            "default": "120",
+            "minimum": "0"
+          }
         }
       ]
     },
diff --git a/widgets/BigQueryTable-batchsink.json b/widgets/BigQueryTable-batchsink.json
@@ -236,6 +236,15 @@
             },
             "default": "false"
           }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Read Timeout",
+          "name": "readTimeout",
+          "widget-attributes": {
+            "default": "120",
+            "minimum": "0"
+          }
         }
       ]
     },
diff --git a/widgets/BigQueryTable-batchsource.json b/widgets/BigQueryTable-batchsource.json
@@ -166,6 +166,15 @@
           "widget-attributes": {
             "placeholder": "projects/<gcp-project-id>/locations/<key-location>/keyRings/<key-ring-name>/cryptoKeys/<key-name>"
           }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Read Timeout",
+          "name": "readTimeout",
+          "widget-attributes": {
+            "default": "120",
+            "minimum": "0"
+          }
         }
       ]
     },

Original file line number	Diff line number	Diff line change
`@@ -417,7 +417,7 @@ private void validatePartitionProperties(@Nullable Schema schema, FailureCollect`
`417`	`417`	`}`
`418`	`418`
`419`	`419`	`Table table = BigQueryUtil.getBigQueryTable(project, dataset, tableName, serviceAccount,`
`420`		`- isServiceAccountFilePath(), collector, null);`
	`420`	`+ isServiceAccountFilePath(), collector, getReadTimeout());`
`421`	`421`	`if (table != null) {`
`422`	`422`	`StandardTableDefinition tableDefinition = table.getDefinition();`
`423`	`423`	`TimePartitioning timePartitioning = tableDefinition.getTimePartitioning();`