Merge pull request #60 from cloudsufi/sheets-filetype-check-dev

[PLUGIN-1817] Added check for non-spreadsheet files with proper error.

src/main/java/io/cdap/plugin/google/common/APIRequestRetryer.java

@@ -51,9 +51,9 @@ public abstract class APIRequestRetryer {
   protected static final int LIMIT_RATE_EXCEEDED_CODE = 403;
   protected static final int BACKEND_ERROR_CODE = 500;
   protected static final int SERVICE_UNAVAILABLE_CODE = 503;
-  private static final int MAX_RETRY_WAIT = 200;
-  private static final int MAX_RETRY_COUNT = 8;
-  private static final int MAX_RETRY_JITTER_WAIT = 100;
+  private static final int MAX_RETRY_WAIT_SECONDS = 300;
+  private static final int MAX_RETRY_COUNT = 10;
+  private static final int MAX_RETRY_JITTER_WAIT_MS = 100;
   protected static final String TOO_MANY_REQUESTS_MESSAGE = "Too Many Requests";
   protected static final String LIMIT_RATE_EXCEEDED_MESSAGE = "Rate Limit Exceeded";
   protected static final String FORBIDDEN_STATUS_MESSAGE = "Forbidden";
@@ -76,7 +76,7 @@ public <V> void onRetry(Attempt<V> attempt) {
             GoogleJsonResponseException e = (GoogleJsonResponseException) exceptionCause;
             LOG.warn(String.format(
               "Error code: '%d', message: '%s'. Attempt: '%d'. Delay since first: '%d'. Description: '%s'.",
-              e.getDetails().getCode(),
+              e.getStatusCode(),
               e.getStatusMessage(),
               attempt.getAttemptNumber(),
               attempt.getDelaySinceFirstAttempt(),
@@ -96,8 +96,8 @@ public <V> void onRetry(Attempt<V> attempt) {
       .retryIfException(APIRequestRetryer::checkThrowable)
       .retryIfExceptionOfType(SocketTimeoutException.class)
       .withWaitStrategy(WaitStrategies.join(
-        new TrueExponentialWaitStrategy(1000, TimeUnit.SECONDS.toMillis(MAX_RETRY_WAIT)),
-        WaitStrategies.randomWait(MAX_RETRY_JITTER_WAIT, TimeUnit.MILLISECONDS)))
+        new TrueExponentialWaitStrategy(1000, TimeUnit.SECONDS.toMillis(MAX_RETRY_WAIT_SECONDS)),
+        WaitStrategies.randomWait(MAX_RETRY_JITTER_WAIT_MS, TimeUnit.MILLISECONDS)))
       .withStopStrategy(StopStrategies.stopAfterAttempt(MAX_RETRY_COUNT))
       .withRetryListener(listener)
       .build();
@@ -125,21 +125,22 @@ private static boolean checkHttpResponseException(Throwable t) {
 
   private static boolean isTooManyRequestsError(GoogleJsonResponseException e) {
     List<String> possibleMessages = Arrays.asList(TOO_MANY_REQUESTS_MESSAGE, LIMIT_RATE_EXCEEDED_MESSAGE);
-    return e.getDetails().getCode() == TOO_MANY_REQUESTS_CODE && possibleMessages.contains(e.getStatusMessage());
+    return e.getStatusCode() == TOO_MANY_REQUESTS_CODE && possibleMessages.contains(
+      e.getStatusMessage());
   }
 
   private static boolean isRateLimitError(GoogleJsonResponseException e) {
-    return e.getDetails().getCode() == LIMIT_RATE_EXCEEDED_CODE
+    return e.getStatusCode() == LIMIT_RATE_EXCEEDED_CODE
       && (LIMIT_RATE_EXCEEDED_MESSAGE.equals(e.getStatusMessage())
       || e.getDetails().getMessage().contains(LIMIT_RATE_EXCEEDED_MESSAGE));
   }
 
   private static boolean isBackendError(GoogleJsonResponseException e) {
-    return e.getDetails().getCode() == BACKEND_ERROR_CODE;
+    return e.getStatusCode() == BACKEND_ERROR_CODE;
   }
 
   private static boolean isServiceUnavailableError(GoogleJsonResponseException e) {
-    return e.getDetails().getCode() == SERVICE_UNAVAILABLE_CODE;
+    return e.getStatusCode() == SERVICE_UNAVAILABLE_CODE;
   }
 
   private static boolean isRateLimitError(HttpResponseException e) {

src/main/java/io/cdap/plugin/google/common/GoogleDriveFilteringClient.java

@@ -72,9 +72,10 @@ public List<File> getFilesSummary(List<ExportedType> exportedTypes, int filesNum
       int retrievedFiles = 0;
       int actualFilesNumber = filesNumber;
       if (IdentifierType.FILE_IDENTIFIER.equals(config.getIdentifierType())) {
-        files.add(service.files().get(config.getFileIdentifier()).setSupportsAllDrives(true).execute());
+        files.add(getFilesSummaryByFileId());
         return files;
       }
+
       Drive.Files.List request = service.files().list()
         .setSupportsAllDrives(true)
         .setIncludeItemsFromAllDrives(true)
@@ -99,6 +100,10 @@ public List<File> getFilesSummary(List<ExportedType> exportedTypes, int filesNum
     });
   }
 
+  protected File getFilesSummaryByFileId() throws IOException, ExecutionException {
+    return service.files().get(config.getFileIdentifier()).setSupportsAllDrives(true).execute();
+  }
+
   private String generateFilter(List<ExportedType> exportedTypes) throws InterruptedException {
     StringBuilder sb = new StringBuilder();
 

src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsFilteringClient.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright © 2024 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.google.sheets.source;
+
+import com.google.api.services.drive.model.File;
+import io.cdap.plugin.google.common.GoogleDriveFilteringClient;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Client for getting File information via Google Sheets API.
+ */
+public class GoogleSheetsFilteringClient extends GoogleDriveFilteringClient<GoogleSheetsSourceConfig> {
+
+  public GoogleSheetsFilteringClient(GoogleSheetsSourceConfig config) throws IOException {
+    super(config);
+  }
+
+  @Override
+  protected File getFilesSummaryByFileId() throws IOException, ExecutionException {
+    File file = service.files().get(config.getFileIdentifier()).setSupportsAllDrives(true).execute();
+    if (!file.getMimeType().equalsIgnoreCase(DRIVE_SPREADSHEETS_MIME)) {
+      throw new ExecutionException(
+        String.format("File with id: '%s' has a MIME_TYPE '%s' and is not a Google Sheets File.",
+                      file.getMimeType(),
+                      config.getFileIdentifier()), null);
+    }
+    return file;
+  }
+}

src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsInputFormat.java

@@ -57,7 +57,7 @@ public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
         GoogleSheetsInputFormatProvider.GSON.fromJson(headersJson, headersType);
 
     // get all sheets files according to filter
-    GoogleDriveFilteringClient driveFilteringClient = new GoogleDriveFilteringClient(googleSheetsSourceConfig);
+    GoogleDriveFilteringClient driveFilteringClient = new GoogleSheetsFilteringClient(googleSheetsSourceConfig);
     List<File> spreadsheetsFiles;
     try {
       spreadsheetsFiles = driveFilteringClient.getFilesSummary(Collections.singletonList(ExportedType.SPREADSHEETS));

src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSourceConfig.java

@@ -334,7 +334,7 @@ public ValidationResult validate(FailureCollector collector) {
       GoogleDriveFilteringClient driveClient;
       GoogleSheetsSourceClient sheetsSourceClient;
       try {
-        driveClient = new GoogleDriveFilteringClient(this);
+        driveClient = new GoogleSheetsFilteringClient(this);
         sheetsSourceClient = new GoogleSheetsSourceClient(this);
       } catch (IOException e) {
         collector.addFailure("Exception during drive and sheets connections instantiating.", null);

src/main/java/io/cdap/plugin/google/sheets/source/SheetTransformer.java

@@ -69,7 +69,8 @@ public static StructuredRecord transform(RowRecord rowRecord, Schema schema, boo
         builder.set(metadataRecordName, rowRecord.getMetadata());
       } else {
         ComplexSingleValueColumn complexSingleValueColumn = rowRecord.getHeaderedCells().get(name);
-        if (complexSingleValueColumn.getData() == null && complexSingleValueColumn.getSubColumns().isEmpty()) {
+        if (complexSingleValueColumn == null || (complexSingleValueColumn.getData() == null
+          && complexSingleValueColumn.getSubColumns().isEmpty())) {
           builder.set(name, null);
         } else {
           processCellData(builder, field, complexSingleValueColumn);

src/test/java/io/cdap/plugin/google/common/APIRequestRetryerTest.java

@@ -33,14 +33,12 @@
 @PrepareForTest({GoogleJsonResponseException.class})
 public class APIRequestRetryerTest {
   private static final int UNPROCESSED_CODE = 504;
-  private static final int RETRY_NUMBER = 8;
+  private static final int RETRY_NUMBER = 10;
 
   @Test
   public void testRetryCount() throws ExecutionException {
     GoogleJsonResponseException exception = PowerMockito.mock(GoogleJsonResponseException.class);
-    GoogleJsonError googleJsonError = new GoogleJsonError();
-    googleJsonError.setCode(APIRequestRetryer.TOO_MANY_REQUESTS_CODE);
-    PowerMockito.when(exception.getDetails()).thenReturn(googleJsonError);
+    PowerMockito.when(exception.getStatusCode()).thenReturn(APIRequestRetryer.TOO_MANY_REQUESTS_CODE);
     PowerMockito.when(exception.getStatusMessage()).thenReturn(APIRequestRetryer.TOO_MANY_REQUESTS_MESSAGE);