Merge pull request #73 from cloudsufi/patch/code-refactor

psainics · web-flow · commit 66c341f0c06d · 2025-08-19T08:15:04.000+05:30
Code Refactor
diff --git a/src/main/java/io/cdap/plugin/google/common/GoogleAuthBaseConfig.java b/src/main/java/io/cdap/plugin/google/common/GoogleAuthBaseConfig.java
@@ -156,7 +156,7 @@ public abstract class GoogleAuthBaseConfig extends PluginConfig {
    * @param collector the failure collector is provided
    * @return The ValidationResult
    */
-  public ValidationResult validate(FailureCollector collector) {
+  public ValidationResult getValidationResult(FailureCollector collector) {
     IdUtils.validateReferenceName(referenceName, collector);
     checkIfDirectoryOrFileIdentifierExists(collector);
     ValidationResult validationResult = new ValidationResult();
@@ -171,7 +171,7 @@ public ValidationResult validate(FailureCollector collector) {
           propertiesAreValid = validateServiceAccount(collector);
           break;
         default:
-          collector.addFailure(String.format("'%s' is not processed value.", authType.toString()), null)
+          collector.addFailure(String.format("'%s' is not processed value.", authType), null)
             .withConfigProperty(AUTH_TYPE);
           return validationResult;
       }
@@ -240,7 +240,7 @@ private boolean validateServiceAccount(FailureCollector collector) {
           .withConfigProperty(NAME_SERVICE_ACCOUNT_JSON);
       }
     }
-    return collector.getValidationFailures().size() == 0;
+    return collector.getValidationFailures().isEmpty();
   }
 
   private boolean isDirectoryOrFileAccessible(FailureCollector collector, GoogleDriveClient driveClient)
diff --git a/src/main/java/io/cdap/plugin/google/common/GoogleFilteringSourceConfig.java b/src/main/java/io/cdap/plugin/google/common/GoogleFilteringSourceConfig.java
@@ -82,8 +82,8 @@ public GoogleFilteringSourceConfig() {
    * @param collector the failure collector is provided
    * @return The ValidationResult
    */
-  public ValidationResult validate(FailureCollector collector) {
-    ValidationResult validationResult = super.validate(collector);
+  public ValidationResult getValidationResult(FailureCollector collector) {
+    ValidationResult validationResult = super.getValidationResult(collector);
     if (validateModificationDateRange(collector)
       && getModificationDateRangeType().equals(ModifiedDateRangeType.CUSTOM)) {
       if (checkPropertyIsSet(collector, startDate, START_DATE, START_DATE_LABEL)) {
diff --git a/src/main/java/io/cdap/plugin/google/drive/sink/GoogleDriveSinkConfig.java b/src/main/java/io/cdap/plugin/google/drive/sink/GoogleDriveSinkConfig.java
@@ -60,7 +60,8 @@ public class GoogleDriveSinkConfig extends GoogleInputSchemaFieldsUsageConfig {
    * @param schema the schema to check compatibility
    */
   public void validate(FailureCollector collector, Schema schema) {
-    super.validate(collector);
+    super.getValidationResult(collector);
+    collector.getOrThrowException();
 
     // validate body field is in schema and has valid format
     validateSchemaField(collector, schema, SCHEMA_BODY_FIELD_NAME, schemaBodyFieldName,
diff --git a/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveInputFormat.java b/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveInputFormat.java
@@ -18,7 +18,9 @@
 
 import com.github.rholder.retry.RetryException;
 import com.google.api.services.drive.model.File;
+import io.cdap.cdap.api.data.format.StructuredRecord;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -33,7 +35,7 @@
 /**
  * Input format class which generates splits for each query.
  */
-public class GoogleDriveInputFormat extends InputFormat {
+public class GoogleDriveInputFormat extends InputFormat<NullWritable, StructuredRecord> {
 
   @Override
   public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
@@ -75,7 +77,8 @@ private GoogleDriveSplit getSplitWithUnlimitedPartitionSize(String fileId) {
   }
 
   @Override
-  public RecordReader createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) {
+  public RecordReader<NullWritable, StructuredRecord> createRecordReader(InputSplit inputSplit,
+      TaskAttemptContext taskAttemptContext) {
     return new GoogleDriveRecordReader();
   }
 }
diff --git a/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveRecordReader.java b/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveRecordReader.java
@@ -17,7 +17,7 @@
 package io.cdap.plugin.google.drive.source;
 
 import com.github.rholder.retry.RetryException;
-import io.cdap.plugin.google.drive.common.FileFromFolder;
+import io.cdap.cdap.api.data.format.StructuredRecord;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputSplit;
@@ -28,12 +28,13 @@
 import java.util.concurrent.ExecutionException;
 
 /**
- * RecordReader implementation, which reads {@link FileFromFolder} wrappers from Google Drive using
+ * RecordReader implementation, which reads {@link StructuredRecord} wrappers from Google Drive using
  * Google Drive API.
  */
-public class GoogleDriveRecordReader extends RecordReader<NullWritable, FileFromFolder> {
+public class GoogleDriveRecordReader extends RecordReader<NullWritable, StructuredRecord> {
 
   private GoogleDriveSourceClient googleDriveSourceClient;
+  private GoogleDriveSourceConfig googleDriveSourceConfig;
   private String fileId;
   private long bytesFrom;
   private long bytesTo;
@@ -43,6 +44,7 @@ public class GoogleDriveRecordReader extends RecordReader<NullWritable, FileFrom
   @Override
   public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException {
     Configuration conf = taskAttemptContext.getConfiguration();
+    googleDriveSourceConfig = GoogleDriveInputFormatProvider.extractPropertiesFromConfig(conf);
     GoogleDriveSourceConfig googleDriveSourceConfig =
       GoogleDriveInputFormatProvider.extractPropertiesFromConfig(conf);
     googleDriveSourceClient = new GoogleDriveSourceClient(googleDriveSourceConfig);
@@ -62,18 +64,20 @@ public boolean nextKeyValue() {
 
   @Override
   public NullWritable getCurrentKey() {
-    return null;
+    return NullWritable.get();
   }
 
   @Override
-  public FileFromFolder getCurrentValue() throws IOException {
+  public StructuredRecord getCurrentValue() throws IOException {
     // read file and content
     isFileProcessed = true;
     try {
       if (isPartitioned) {
-        return googleDriveSourceClient.getFilePartition(fileId, bytesFrom, bytesTo);
+        return FilesFromFolderTransformer.transform(
+            googleDriveSourceClient.getFilePartition(fileId, bytesFrom, bytesTo), googleDriveSourceConfig.getSchema());
       } else {
-        return googleDriveSourceClient.getFile(fileId);
+        return FilesFromFolderTransformer.transform(googleDriveSourceClient.getFile(fileId),
+            googleDriveSourceConfig.getSchema());
       }
     } catch (ExecutionException | RetryException e) {
       throw new RuntimeException("Exception during file or file part reading.", e);
@@ -88,6 +92,6 @@ public float getProgress() {
 
   @Override
   public void close() {
-
+    // no-op
   }
 }
diff --git a/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveSource.java b/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveSource.java
@@ -41,7 +41,7 @@
 @Plugin(type = BatchSource.PLUGIN_TYPE)
 @Name(GoogleDriveSource.NAME)
 @Description("Reads fileset from specified Google Drive directory.")
-public class GoogleDriveSource extends BatchSource<NullWritable, FileFromFolder, StructuredRecord> {
+public class GoogleDriveSource extends BatchSource<NullWritable, StructuredRecord, StructuredRecord> {
   public static final String NAME = "GoogleDrive";
 
   private final GoogleDriveSourceConfig config;
@@ -53,7 +53,7 @@ public GoogleDriveSource(GoogleDriveSourceConfig config) {
   @Override
   public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     FailureCollector failureCollector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
-    config.validate(failureCollector);
+    config.getValidationResult(failureCollector);
     failureCollector.getOrThrowException();
 
     pipelineConfigurer.getStageConfigurer().setOutputSchema(config.getSchema());
@@ -62,7 +62,7 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
   @Override
   public void prepareRun(BatchSourceContext context) {
     FailureCollector failureCollector = context.getFailureCollector();
-    config.validate(failureCollector);
+    config.getValidationResult(failureCollector);
     failureCollector.getOrThrowException();
 
     LineageRecorder lineageRecorder = new LineageRecorder(context, config.getReferenceName());
@@ -74,9 +74,4 @@ public void prepareRun(BatchSourceContext context) {
 
     context.setInput(Input.of(config.getReferenceName(), new GoogleDriveInputFormatProvider(config)));
   }
-
-  @Override
-  public void transform(KeyValue<NullWritable, FileFromFolder> input, Emitter<StructuredRecord> emitter) {
-    emitter.emit(FilesFromFolderTransformer.transform(input.getValue(), config.getSchema()));
-  }
 }
diff --git a/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveSourceConfig.java b/src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveSourceConfig.java
@@ -136,8 +136,8 @@ public Schema getSchema() {
    * @param collector the failure collector is provided
    * @return The ValidationResult
    */
-  public ValidationResult validate(FailureCollector collector) {
-    ValidationResult validationResult = super.validate(collector);
+  public ValidationResult getValidationResult(FailureCollector collector) {
+    ValidationResult validationResult = super.getValidationResult(collector);
 
     validateFileTypesToPull(collector);
 
diff --git a/src/main/java/io/cdap/plugin/google/sheets/sink/GoogleSheetsSinkConfig.java b/src/main/java/io/cdap/plugin/google/sheets/sink/GoogleSheetsSinkConfig.java
@@ -167,7 +167,8 @@ public class GoogleSheetsSinkConfig extends GoogleInputSchemaFieldsUsageConfig {
    * @param schema the schema to check compatibility
    */
   public void validate(FailureCollector collector, Schema schema) {
-    super.validate(collector);
+    super.getValidationResult(collector);
+    collector.getOrThrowException();
 
     // validate spreadsheet name field is in schema and has valid format
     validateSchemaField(collector, schema, SCHEMA_SPREAD_SHEET_NAME_FIELD_NAME, schemaSpreadsheetNameFieldName,
diff --git a/src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSource.java b/src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSource.java
@@ -54,7 +54,7 @@ public GoogleSheetsSource(GoogleSheetsSourceConfig config) {
   public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer();
     FailureCollector failureCollector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
-    config.validate(failureCollector);
+    config.getValidationResult(failureCollector);
     failureCollector.getOrThrowException();
 
     Schema configuredSchema = config.getSchema(failureCollector);
@@ -64,7 +64,7 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
   @Override
   public void prepareRun(BatchSourceContext context) {
     FailureCollector failureCollector = context.getFailureCollector();
-    config.validate(failureCollector);
+    config.getValidationResult(failureCollector);
     failureCollector.getOrThrowException();
 
     Schema configSchema = config.getSchema(failureCollector);
diff --git a/src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSourceConfig.java b/src/main/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSourceConfig.java
@@ -318,8 +318,8 @@ private boolean shouldGetSchema() {
    * @param collector the failure collector is provided
    * @return The ValidationResult
    */
-  public ValidationResult validate(FailureCollector collector) {
-    ValidationResult validationResult = super.validate(collector);
+  public ValidationResult getValidationResult(FailureCollector collector) {
+    ValidationResult validationResult = super.getValidationResult(collector);
 
     // reset current headers info
     dataSchemaInfo = new LinkedHashMap<>();
diff --git a/src/test/java/io/cdap/plugin/google/common/GoogleAuthBaseConfigTest.java b/src/test/java/io/cdap/plugin/google/common/GoogleAuthBaseConfigTest.java
@@ -60,7 +60,7 @@ public void testValidationErrorFilePath() {
     config.setIdentifierType(IdentifierType.FILE_IDENTIFIER.name());
     config.setFileIdentifier("fileId");
     FailureCollector collector = new DefaultFailureCollector("stageConfig", Collections.EMPTY_MAP);
-    config.validate(collector);
+    config.getValidationResult(collector);
     Assert.assertEquals(1, collector.getValidationFailures().size());
     Assert.assertEquals("Service Account File Path is not available.",
                         collector.getValidationFailures().get(0).getMessage());
@@ -81,7 +81,7 @@ public void testValidationErrorJSON() {
     config.setFileIdentifier("fileId");
     config.setIdentifierType(IdentifierType.FILE_IDENTIFIER.name());
     FailureCollector collector = new DefaultFailureCollector("stageConfig", Collections.EMPTY_MAP);
-    config.validate(collector);
+    config.getValidationResult(collector);
     Assert.assertEquals(1, collector.getValidationFailures().size());
     Assert.assertEquals("Service Account JSON can not be empty.",
                         collector.getValidationFailures().get(0).getMessage());
@@ -101,7 +101,7 @@ public void testValidationOauthWithoutAccessToken() {
     config.setIdentifierType(IdentifierType.FILE_IDENTIFIER.name());
     config.setFileIdentifier("fileId");
     FailureCollector collector = new DefaultFailureCollector("stageConfig", Collections.EMPTY_MAP);
-    config.validate(collector);
+    config.getValidationResult(collector);
     Assert.assertEquals(1, collector.getValidationFailures().size());
     Assert.assertEquals("'Access Token' property is empty or macro is not available.",
                         collector.getValidationFailures().get(0).getMessage());
@@ -121,7 +121,7 @@ public void testValidationOauthWithoutRefreshToken() {
     config.setFileIdentifier("fileId");
     config.setIdentifierType(IdentifierType.FILE_IDENTIFIER.name());
     FailureCollector collector = new DefaultFailureCollector("stageConfig", Collections.EMPTY_MAP);
-    config.validate(collector);
+    config.getValidationResult(collector);
     Assert.assertEquals(3, collector.getValidationFailures().size());
     Assert.assertEquals("'Client ID' property is empty or macro is not available.",
                         collector.getValidationFailures().get(0).getMessage());
@@ -148,7 +148,7 @@ public void testWithDirectoryIdAsNull() {
     config.setAccessToken("access");
     config.setoAuthMethod(OAuthMethod.ACCESS_TOKEN.name());
     FailureCollector collector = new DefaultFailureCollector("stageConfig", Collections.EMPTY_MAP);
-    config.validate(collector);
+    config.getValidationResult(collector);
     Assert.assertEquals(2, collector.getValidationFailures().size());
     Assert.assertEquals("Directory Identifier can not be null.",
                         collector.getValidationFailures().get(0).getMessage());
@@ -171,7 +171,7 @@ public void testWithFileIdAsNull() {
     config.setIdentifierType(IdentifierType.FILE_IDENTIFIER.name());
     config.setoAuthMethod(OAuthMethod.ACCESS_TOKEN.name());
     FailureCollector collector = new DefaultFailureCollector("stageConfig", Collections.EMPTY_MAP);
-    config.validate(collector);
+    config.getValidationResult(collector);
     Assert.assertEquals(2, collector.getValidationFailures().size());
     Assert.assertEquals("File Identifier can not be null.",
                         collector.getValidationFailures().get(0).getMessage());
diff --git a/src/test/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSourceConfigTest.java b/src/test/java/io/cdap/plugin/google/sheets/source/GoogleSheetsSourceConfigTest.java
@@ -104,7 +104,7 @@ public void testGetMetadataCoordinates() throws NoSuchFieldException, IllegalAcc
   }
 
   @Test
-  public void testValidateMetadataCellsOnlyHeader() throws NoSuchFieldException, IllegalAccessException,
+  public void testGetValidationResultMetadataCellsOnlyHeader() throws NoSuchFieldException, IllegalAccessException,
     NoSuchMethodException, InvocationTargetException {
     Method validateMetadataCellsMethod =
       config.getClass().getDeclaredMethod("validateMetadataCells", FailureCollector.class);
@@ -141,7 +141,7 @@ public void testValidateMetadataCellsOnlyHeader() throws NoSuchFieldException, I
   }
 
   @Test
-  public void testValidateMetadataCellsOnlyFooter() throws NoSuchFieldException, IllegalAccessException,
+  public void testGetValidationResultMetadataCellsOnlyFooter() throws NoSuchFieldException, IllegalAccessException,
     NoSuchMethodException, InvocationTargetException {
     Method validateMetadataCellsMethod =
       config.getClass().getDeclaredMethod("validateMetadataCells", FailureCollector.class);
@@ -180,7 +180,7 @@ public void testValidateMetadataCellsOnlyFooter() throws NoSuchFieldException, I
   }
 
   @Test
-  public void testValidateMetadataCellsHeaderAndFooter() throws NoSuchFieldException, IllegalAccessException,
+  public void testGetValidationResultMetadataCellsHeaderAndFooter() throws NoSuchFieldException, IllegalAccessException,
     NoSuchMethodException, InvocationTargetException {
     Method validateMetadataCellsMethod =
       config.getClass().getDeclaredMethod("validateMetadataCells", FailureCollector.class);

Original file line number	Diff line number	Diff line change
`@@ -156,7 +156,7 @@ public abstract class GoogleAuthBaseConfig extends PluginConfig {`
`156`	`156`	`* @param collector the failure collector is provided`
`157`	`157`	`* @return The ValidationResult`
`158`	`158`	`*/`
`159`		`- public ValidationResult validate(FailureCollector collector) {`
	`159`	`+ public ValidationResult getValidationResult(FailureCollector collector) {`
`160`	`160`	`IdUtils.validateReferenceName(referenceName, collector);`
`161`	`161`	`checkIfDirectoryOrFileIdentifierExists(collector);`
`162`	`162`	`ValidationResult validationResult = new ValidationResult();`
`@@ -171,7 +171,7 @@ public ValidationResult validate(FailureCollector collector) {`
`171`	`171`	`propertiesAreValid = validateServiceAccount(collector);`
`172`	`172`	`break;`
`173`	`173`	`default:`
`174`		`- collector.addFailure(String.format("'%s' is not processed value.", authType.toString()), null)`
	`174`	`+ collector.addFailure(String.format("'%s' is not processed value.", authType), null)`
`175`	`175`	`.withConfigProperty(AUTH_TYPE);`
`176`	`176`	`return validationResult;`
`177`	`177`	`}`
`@@ -240,7 +240,7 @@ private boolean validateServiceAccount(FailureCollector collector) {`
`240`	`240`	`.withConfigProperty(NAME_SERVICE_ACCOUNT_JSON);`
`241`	`241`	`}`
`242`	`242`	`}`
`243`		`- return collector.getValidationFailures().size() == 0;`
	`243`	`+ return collector.getValidationFailures().isEmpty();`
`244`	`244`	`}`
`245`	`245`
`246`	`246`	`private boolean isDirectoryOrFileAccessible(FailureCollector collector, GoogleDriveClient driveClient)`