Improve read session expired error message

bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/BigQueryUtil.java

@@ -47,6 +47,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Multimap;
 import io.grpc.Status;
+import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -82,6 +83,9 @@ public class BigQueryUtil {
           "HTTP/2 error code: INTERNAL_ERROR",
           "Connection closed with unknown cause",
           "Received unexpected EOS on DATA frame from server");
+
+  static final String READ_SESSION_EXPIRED_ERROR_MESSAGE = "session expired at";
+
   private static final String PROJECT_PATTERN = "\\S+";
   private static final String DATASET_PATTERN = "\\w+";
   // Allow all non-whitespace beside ':' and '.'.
@@ -118,6 +122,19 @@ static boolean isRetryableInternalError(Throwable t) {
     return false;
   }
 
+  public static boolean isReadSessionExpired(Throwable cause) {
+    return getCausalChain(cause).stream().anyMatch(BigQueryUtil::isReadSessionExpiredInternalError);
+  }
+
+  static boolean isReadSessionExpiredInternalError(Throwable t) {
+    if (t instanceof StatusRuntimeException) {
+      StatusRuntimeException statusRuntimeException = (StatusRuntimeException) t;
+      return statusRuntimeException.getStatus().getCode() == Code.FAILED_PRECONDITION
+          && statusRuntimeException.getMessage().contains(READ_SESSION_EXPIRED_ERROR_MESSAGE);
+    }
+    return false;
+  }
+
   static BigQueryException convertToBigQueryException(BigQueryError error) {
     return new BigQueryException(UNKNOWN_CODE, error.getMessage(), error);
   }

bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/ReadSessionCreator.java

@@ -60,7 +60,7 @@ private static synchronized void initializeCache(long readSessionCacheDurationMi
     if (!initialized) {
       READ_SESSION_CACHE =
           CacheBuilder.newBuilder()
-              .expireAfterWrite(readSessionCacheDurationMins, TimeUnit.MINUTES)
+              .expireAfterWrite(600, TimeUnit.MINUTES)
               .maximumSize(1000)
               .build();
       initialized = true;

bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/StreamCombiningIterator.java

@@ -311,7 +311,19 @@ public void onError(Throwable t) {
         newConnection(this, builder);
         retries++;
       } else {
-        stopWithError(t);
+        if (BigQueryUtil.isReadSessionExpired(t)) {
+          RuntimeException runtimeException =
+              new RuntimeException(
+                  "Read session expired after 6 hours. Dataframes on BigQuery tables cannot be "
+                      + "operated on beyond this time-limit. Your options: "
+                      + "(1) try finishing within 6 hours, "
+                      + "(2) use Spark dataframe caching, "
+                      + "(3) read from a newly created BigQuery dataframe.",
+                  t);
+          stopWithError(runtimeException);
+        } else {
+          stopWithError(t);
+        }
       }
     }