diff --git a/src/main/java/io/cdap/e2e/pages/actions/CdfBigQueryPropertiesActions.java b/src/main/java/io/cdap/e2e/pages/actions/CdfBigQueryPropertiesActions.java
index 032fb90b2..2f848d480 100644
--- a/src/main/java/io/cdap/e2e/pages/actions/CdfBigQueryPropertiesActions.java
+++ b/src/main/java/io/cdap/e2e/pages/actions/CdfBigQueryPropertiesActions.java
@@ -37,6 +37,11 @@ public class CdfBigQueryPropertiesActions {
     cdfBigQueryPropertiesLocators = SeleniumHelper.getPropertiesLocators(CdfBigQueryPropertiesLocators.class);
   }
 
+  /**
+   * @deprecated
+   * Call individual actions as per test scenario in step design file.
+   */
+  @Deprecated
   public static void enterBigQueryProperties(String tableProp) throws InterruptedException, IOException {
     CdfStudioLocators.bigQueryProperties.click();
     CdfBigQueryPropertiesLocators.bigQueryReferenceName.sendKeys(AUTOMATION_TEST);
@@ -51,9 +56,8 @@ public static void enterBigQueryProperties(String tableProp) throws InterruptedE
     SeleniumHelper.waitElementIsVisible(CdfBigQueryPropertiesLocators.textSuccess, ONE);
   }
 
-  public static void enterCmekProperty(String prop) throws IOException {
-    CdfStudioLocators.bigQueryProperties.click();
-    CdfBigQueryPropertiesLocators.cmekKey.sendKeys(SeleniumHelper.readParameters(prop));
+  public static void enterCmekProperty(String cmek) throws IOException {
+    CdfBigQueryPropertiesLocators.cmekKey.sendKeys(cmek);
   }
 
   public static void enterFilePath(String path) throws InterruptedException, IOException {
@@ -120,6 +124,12 @@ public static void enterTemporaryBucketName(String bucket) throws IOException {
     SeleniumHelper.replaceElementValue(CdfBigQueryPropertiesLocators.temporaryBucketName, bucket);
   }
 
+  /**
+   * @deprecated
+   * Use either {@link io.cdap.e2e.utils.CdfHelper#openSinkPluginPreviewData(String)}
+   * or {@link io.cdap.e2e.utils.CdfHelper#openSourcePluginPreviewData(String)} as per plugin type.
+   */
+  @Deprecated
   public static void clickPreviewData() {
     SeleniumHelper.waitAndClick(CdfBigQueryPropertiesLocators.previewData);
   }
diff --git a/src/main/java/io/cdap/e2e/pages/actions/CdfGcsActions.java b/src/main/java/io/cdap/e2e/pages/actions/CdfGcsActions.java
index d27445719..e2250aaaa 100644
--- a/src/main/java/io/cdap/e2e/pages/actions/CdfGcsActions.java
+++ b/src/main/java/io/cdap/e2e/pages/actions/CdfGcsActions.java
@@ -23,6 +23,7 @@
 import org.openqa.selenium.By;
 import org.openqa.selenium.WebElement;
 import java.io.IOException;
+import java.util.Map;
 import java.util.UUID;
 
 import static io.cdap.e2e.utils.ConstantsUtil.DELIMITER;
@@ -65,6 +66,11 @@ public static void enterFormat() {
     SeleniumDriver.getDriver().findElement(By.xpath("//*[contains(text(),'csv')]")).click();
   }
 
+  /**
+   * @deprecated
+   * Use {@link CdfGcsActions#enterSampleSize(String)}
+   */
+  @Deprecated
   public static void enterSamplesize() {
     CdfGCSLocators.samplesize.sendKeys(SAMPLE_SIZE);
   }
@@ -73,6 +79,12 @@ public static void closeButton() {
     CdfGCSLocators.closeButton.click();
   }
 
+  /**
+   * @deprecated
+   * Use either {@link io.cdap.e2e.utils.CdfHelper#openSinkPluginProperties(String)}
+   * or {@link io.cdap.e2e.utils.CdfHelper#openSourcePluginProperties(String)} as per plugin type.
+   */
+  @Deprecated
   public static void gcsProperties() {
     CdfGCSLocators.gcsProperties.click();
   }
@@ -85,6 +97,11 @@ public static void getSchema() {
     CdfGCSLocators.getSchemaButton.click();
   }
 
+  /**
+   * @deprecated
+   * Use {@link CdfGcsActions#enterDelimiterField(String)}
+   */
+  @Deprecated
   public static void delimiter() throws IOException {
     CdfGCSLocators.delimiter.sendKeys(SeleniumHelper.readParameters(DELIMITER));
   }
@@ -95,6 +112,11 @@ public static void selectFormat(String formatType) throws InterruptedException {
       "//li[@data-value='" + formatType + "']")));
   }
 
+  /**
+   * @deprecated
+   * Use {@link io.cdap.e2e.utils.CdfHelper#validateSchema(Map)}
+   */
+  @Deprecated
   public static void validateSchema() {
     for (WebElement schema : CdfGCSLocators.schemaSection) {
       String actualStr = schema.getAttribute(VALUE);
@@ -109,7 +131,7 @@ public static void clickValidateButton() {
   }
 
   public static void validateSuccessMessage() {
-    Assert.assertTrue(CdfGCSLocators.successMessage.isDisplayed());
+    Assert.assertTrue(CdfGCSLocators.validationSuccessMessage.isDisplayed());
   }
 
   public static void selectFileEncoding(String encoding) {
@@ -147,10 +169,46 @@ public static void enterDelimiterField(String delimiter) {
     CdfGCSLocators.delimiterField.sendKeys(delimiter);
   }
 
+  /**
+   * @deprecated
+   * Use either {@link io.cdap.e2e.utils.CdfHelper#openSinkPluginPreviewData(String)}
+   * or {@link io.cdap.e2e.utils.CdfHelper#openSourcePluginPreviewData(String)} as per plugin type.
+   */
+  @Deprecated
   public static void clickPreviewData() {
     SeleniumHelper.waitAndClick(CdfGCSLocators.gcsPreviewData);
   }
   public static void enterSampleSize(String sampleSize) {
     SeleniumHelper.replaceElementValue(CdfGCSLocators.samplesize, sampleSize);
   }
+
+  public static void enterPathSuffix(String pathSuffix) {
+    SeleniumHelper.replaceElementValue(CdfGCSLocators.pathSuffix, pathSuffix);
+  }
+
+  public static void toggleWriteHeader() {
+    CdfGCSLocators.writeHeaderSwitch.click();
+  }
+
+  public static void enterLocation(String location) {
+    SeleniumHelper.replaceElementValue(CdfGCSLocators.location, location);
+  }
+
+  public static void selectContentType(String contentType) {
+    CdfGCSLocators.contentType.click();
+    SeleniumHelper.waitAndClick(SeleniumDriver.getDriver().findElement(By.xpath(
+      "//li[@data-value='" + contentType + "']")));
+  }
+
+  public static void enterEncryptionKeyName(String cmek) {
+    CdfGCSLocators.cmekKey.sendKeys(cmek);
+  }
+
+  public static void enterOutputFilePrefix(String filePrefix) {
+    CdfGCSLocators.outputFilePrefix.sendKeys(filePrefix);
+  }
+
+  public static void enterFileSystemProperties(String fileSystemProperties) {
+    CdfGCSLocators.fileSystemProperties.sendKeys(fileSystemProperties);
+  }
 }
diff --git a/src/main/java/io/cdap/e2e/pages/actions/CdfPipelineRunAction.java b/src/main/java/io/cdap/e2e/pages/actions/CdfPipelineRunAction.java
index 162c6c6f3..abb8e6617 100644
--- a/src/main/java/io/cdap/e2e/pages/actions/CdfPipelineRunAction.java
+++ b/src/main/java/io/cdap/e2e/pages/actions/CdfPipelineRunAction.java
@@ -88,7 +88,4 @@ public static String captureRawLogs() {
     return logs;
   }
 
-  public static void schemaStatusValidation() {
-    Assert.assertTrue(CdfPipelineRunLocators.getSchemaSuccessStatus.isDisplayed());
-  }
 }
diff --git a/src/main/java/io/cdap/e2e/pages/actions/CdfStudioActions.java b/src/main/java/io/cdap/e2e/pages/actions/CdfStudioActions.java
index f713328db..db62dd9df 100644
--- a/src/main/java/io/cdap/e2e/pages/actions/CdfStudioActions.java
+++ b/src/main/java/io/cdap/e2e/pages/actions/CdfStudioActions.java
@@ -36,15 +36,30 @@ public class CdfStudioActions {
     cdfStudioLocators = SeleniumHelper.getPropertiesLocators(CdfStudioLocators.class);
   }
 
+  /**
+   * @deprecated
+   * Use {@link io.cdap.e2e.utils.CdfHelper#selectSourcePlugin(String)}
+   */
+  @Deprecated
   public static void selectGCS() throws InterruptedException {
     WebElement element = CdfStudioLocators.gcsBucket;
     SeleniumHelper.waitAndClick(element);
   }
 
+  /**
+   * @deprecated
+   * Use {@link io.cdap.e2e.utils.CdfHelper#selectSourcePlugin(String)}
+   */
+  @Deprecated
   public static void clickSource() {
     CdfStudioLocators.source.click();
   }
 
+  /**
+   * @deprecated
+   * Use {@link CdfStudioActions#clickSink()} and {@link io.cdap.e2e.utils.CdfHelper#selectSinkPlugin(String)}
+   */
+  @Deprecated
   public static void sinkBigQuery() {
     CdfStudioLocators.sink.click();
     CdfStudioLocators.bigQueryObject.click();
@@ -77,7 +92,7 @@ public static void pipelineDeploy() {
   }
 
   public static String bannerErrorMessage() {
-    return CdfStudioLocators.bannerMssge.getText();
+    return CdfStudioLocators.statusBannerText.getText();
   }
 
   public static void clickSink() {
@@ -88,11 +103,22 @@ public static void clickValidateButton() {
     CdfStudioLocators.validateButton.click();
   }
 
+  /**
+   * @deprecated
+   * Use either {@link io.cdap.e2e.utils.CdfHelper#openSinkPluginProperties(String)}
+   * or {@link io.cdap.e2e.utils.CdfHelper#openSourcePluginProperties(String)} as per plugin type.
+   */
+  @Deprecated
   public static void clickProperties(String plugin) {
     SeleniumDriver.getDriver().findElement(
       By.xpath("//*[contains(@title,'" + plugin + "')]//following-sibling::div")).click();
   }
 
+  /**
+   * @deprecated
+   * Use {@link io.cdap.e2e.utils.CdfHelper#connectSourceAndSink(String, String)}
+   */
+  @Deprecated
   public static void  connectSourceAndSink(String source, String sink) {
     SeleniumHelper.waitElementIsVisible(SeleniumDriver.getDriver().findElement(
       By.xpath("//*[contains(@title,'" + sink + "')]")));
@@ -101,6 +127,11 @@ public static void  connectSourceAndSink(String source, String sink) {
       SeleniumDriver.getDriver().findElement(By.xpath("//*[contains(@title,'" + sink + "')]")));
   }
 
+  /**
+   * @deprecated
+   * Use {@link io.cdap.e2e.utils.CdfHelper#selectSourcePlugin(String)}
+   */
+  @Deprecated
   public static void selectBQ() throws InterruptedException {
     SeleniumHelper.waitAndClick(CdfStudioLocators.sourceBigQuery);
   }
@@ -109,6 +140,11 @@ public static void clickCloseButton() {
     SeleniumHelper.waitAndClick(CdfStudioLocators.closeButton);
   }
 
+  /**
+   * @deprecated
+   * Use {@link CdfStudioActions#clickSink()} and {@link io.cdap.e2e.utils.CdfHelper#selectSinkPlugin(String)}
+   */
+  @Deprecated
   public static void sinkGcs() {
     CdfStudioLocators.sink.click();
     CdfStudioLocators.gcs.click();
diff --git a/src/main/java/io/cdap/e2e/pages/locators/CdfGCSLocators.java b/src/main/java/io/cdap/e2e/pages/locators/CdfGCSLocators.java
index a81d7f0cd..04d3557af 100644
--- a/src/main/java/io/cdap/e2e/pages/locators/CdfGCSLocators.java
+++ b/src/main/java/io/cdap/e2e/pages/locators/CdfGCSLocators.java
@@ -63,8 +63,8 @@ public class CdfGCSLocators {
   @FindBy(how = How.XPATH, using = "//*[@data-cy='plugin-properties-validate-btn']")
   public static WebElement validateBtn;
 
-  @FindBy(how = How.XPATH, using = "//*[@data-cy='plugin-properties-validate-btn']")
-  public static WebElement successMessage;
+  @FindBy(how = How.XPATH, using = "//*[@data-cy='plugin-validation-success-msg']")
+  public static WebElement validationSuccessMessage;
 
   @FindBy(how = How.XPATH, using = "//*[@placeholder='Field name']")
   public static List<WebElement> schemaSection;
@@ -103,4 +103,25 @@ public class CdfGCSLocators {
 
   @FindBy(how = How.XPATH, using = "//*[@data-cy=\"delimiter\" and @class=\"MuiInputBase-input\"]")
   public static WebElement delimiterField;
+
+  @FindBy(how = How.XPATH, using = "//input[@data-cy='suffix']")
+  public static WebElement pathSuffix;
+
+  @FindBy(how = How.XPATH, using = "//*[@data-cy='switch-writeHeader']")
+  public static WebElement writeHeaderSwitch;
+
+  @FindBy(how = How.XPATH, using = "//input[@data-cy='location']")
+  public static WebElement location;
+
+  @FindBy(how = How.XPATH, using = "//*[@data-cy='select-contentType']")
+  public static WebElement contentType;
+
+  @FindBy(how = How.XPATH, using = "//input[@data-cy='cmekKey']")
+  public static WebElement cmekKey;
+
+  @FindBy(how = How.XPATH, using = "//input[@data-cy='outputFileNameBase']")
+  public static WebElement outputFilePrefix;
+
+  @FindBy(how = How.XPATH, using = "//textarea[@data-cy='fileSystemProperties']")
+  public static WebElement fileSystemProperties;
 }
diff --git a/src/main/java/io/cdap/e2e/pages/locators/CdfPipelineRunLocators.java b/src/main/java/io/cdap/e2e/pages/locators/CdfPipelineRunLocators.java
index 427a5d966..e87f0f9ee 100644
--- a/src/main/java/io/cdap/e2e/pages/locators/CdfPipelineRunLocators.java
+++ b/src/main/java/io/cdap/e2e/pages/locators/CdfPipelineRunLocators.java
@@ -71,9 +71,6 @@ public class CdfPipelineRunLocators {
   @FindBy(how = How.XPATH, using = "(//*[contains(@class, 'metric-records-out-label')])[2]/following-sibling::span")
   public static WebElement recordsOutCount;
 
-  @FindBy(how = How.XPATH, using = "//*[@data-cy='plugin-validation-success-msg']")
-  public static WebElement getSchemaSuccessStatus;
-
   @FindBy(how = How.XPATH, using = "//*[contains(text(),'saved successfully.')]")
   public static WebElement savedSuccessMessage;
 
diff --git a/src/main/java/io/cdap/e2e/pages/locators/CdfSchemaLocators.java b/src/main/java/io/cdap/e2e/pages/locators/CdfSchemaLocators.java
new file mode 100644
index 000000000..50cec4c8c
--- /dev/null
+++ b/src/main/java/io/cdap/e2e/pages/locators/CdfSchemaLocators.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright © 2021 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.e2e.pages.locators;
+
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.support.FindBy;
+import org.openqa.selenium.support.How;
+
+import java.util.List;
+
+/**
+ * Represents CdfSchemaLocators
+ */
+public class CdfSchemaLocators {
+  @FindBy(how = How.XPATH,
+    using = "//div[@data-cy='Output Schema']//div[@data-cy='schema-fields-list']//*[@placeholder='Field name']")
+  public static List<WebElement> outputSchemaColumnNames;
+
+  @FindBy(how = How.XPATH, using = "//div[@data-cy='Output Schema']//div[@data-cy='schema-fields-list']//select")
+  public static List<WebElement> outputSchemaDataTypes;
+
+  @FindBy(how = How.XPATH,
+    using = "//div[@data-cy='Input Schema']//div[@data-cy='schema-fields-list']//*[@placeholder='Field name']")
+  public static List<WebElement> inputSchemaColumnNames;
+
+  @FindBy(how = How.XPATH, using = "//div[@data-cy='Input Schema']//div[@data-cy='schema-fields-list']//select")
+  public static List<WebElement> inputSchemaDataTypes;
+
+  @FindBy(how = How.XPATH, using = "//*[@data-cy='get-schema-btn']//span[text()='Get Schema']")
+  public static WebElement getSchemaLoadComplete;
+}
diff --git a/src/main/java/io/cdap/e2e/pages/locators/CdfStudioLocators.java b/src/main/java/io/cdap/e2e/pages/locators/CdfStudioLocators.java
index 45bf2bf66..238891e07 100644
--- a/src/main/java/io/cdap/e2e/pages/locators/CdfStudioLocators.java
+++ b/src/main/java/io/cdap/e2e/pages/locators/CdfStudioLocators.java
@@ -20,6 +20,8 @@
 import org.openqa.selenium.support.FindBy;
 import org.openqa.selenium.support.How;
 
+import java.util.List;
+
 /**
  * Represents CdfStudioLocators
  */
@@ -57,9 +59,6 @@ public class CdfStudioLocators {
   @FindBy(how = How.XPATH, using = "//*[contains(text(),'Preview')]")
   public static WebElement preview;
 
-  @FindBy(how = How.XPATH, using = "//*[@class='ng-binding ng-scope']")
-  public static WebElement bannerMssge;
-
   @FindBy(how = How.XPATH, using = "//*[@class=\"pipeline-name ng-binding ng-scope placeholder\"]")
   public static WebElement pipelineName;
 
@@ -107,4 +106,15 @@ public class CdfStudioLocators {
 
   @FindBy(how = How.XPATH, using = "//*[@title=\"GCS\"]//following-sibling::div")
   public static WebElement gcsProperties;
+
+  @FindBy(how = How.XPATH,
+    using = "(//h2[text()='Input Records']/parent::div//div[not(@data-cy='preview-data-row')]/div[text()!=''])")
+  public static List<WebElement> previewInputRecordColumnNames;
+
+  @FindBy(how = How.XPATH,
+    using = "(//h2[text()='Output Records']/parent::div//div[not(@data-cy='preview-data-row')]/div[text()!=''])")
+  public static List<WebElement> previewOutputRecordColumnNames;
+
+  @FindBy(how = How.XPATH, using = "//*[@role='tablist']/li[contains(text(),'Properties')]")
+  public static WebElement previewPropertiesTab;
 }
diff --git a/src/main/java/io/cdap/e2e/utils/GcpClient.java b/src/main/java/io/cdap/e2e/utils/BigQueryClient.java
similarity index 95%
rename from src/main/java/io/cdap/e2e/utils/GcpClient.java
rename to src/main/java/io/cdap/e2e/utils/BigQueryClient.java
index 3d69aab61..18482ae81 100644
--- a/src/main/java/io/cdap/e2e/utils/GcpClient.java
+++ b/src/main/java/io/cdap/e2e/utils/BigQueryClient.java
@@ -27,9 +27,9 @@
 import java.util.Optional;
 
 /**
- * Represents GcpClient
+ * Represents BigQuery Client
  */
-public class GcpClient {
+public class BigQueryClient {
 
   private static BigQuery bigQueryService = null;
 
@@ -56,7 +56,7 @@ public static void dropBqQuery(String table) throws IOException, InterruptedExce
 
   /**
    * @deprecated
-   * Use {@link GcpClient#getSoleQueryResult(String)} instead and parsing needs to be done by caller.
+   * Use {@link BigQueryClient#getSoleQueryResult(String)} instead and parsing needs to be done by caller.
    */
   @Deprecated
   public static int executeQuery(String query) throws InterruptedException, IOException {
diff --git a/src/main/java/io/cdap/e2e/utils/CdfHelper.java b/src/main/java/io/cdap/e2e/utils/CdfHelper.java
index 54d1228a5..5fabfdd55 100644
--- a/src/main/java/io/cdap/e2e/utils/CdfHelper.java
+++ b/src/main/java/io/cdap/e2e/utils/CdfHelper.java
@@ -22,6 +22,7 @@
 import io.cdap.e2e.pages.actions.CdfStudioActions;
 import io.cdap.e2e.pages.locators.CdfGCSLocators;
 import io.cdap.e2e.pages.locators.CdfPipelineRunLocators;
+import io.cdap.e2e.pages.locators.CdfSchemaLocators;
 import org.junit.Assert;
 import org.openqa.selenium.By;
 import org.openqa.selenium.NoAlertPresentException;
@@ -31,6 +32,8 @@
 import stepsdesign.BeforeActions;
 
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.UUID;
 
 /**
@@ -38,6 +41,8 @@
  */
 public interface CdfHelper {
 
+  CdfSchemaLocators SCHEMA_LOCATORS = SeleniumHelper.getPropertiesLocators(CdfSchemaLocators.class);
+
   default void openCdf() throws IOException, InterruptedException {
     SeleniumDriver.openPage(SeleniumHelper.readParameters(ConstantsUtil.CDFURL));
     try {
@@ -50,7 +55,7 @@ default void openCdf() throws IOException, InterruptedException {
 
   default int countOfNoOfRecordsTransferredToBigQueryIn(String tableName) throws IOException, InterruptedException {
     int countRecords;
-    countRecords = GcpClient.countBqQuery(SeleniumHelper.readParameters(tableName));
+    countRecords = BigQueryClient.countBqQuery(SeleniumHelper.readParameters(tableName));
     BeforeActions.scenario.write("**********No of Records Transferred in table" +
                                    SeleniumHelper.readParameters(tableName) + "*:" + countRecords);
     Assert.assertTrue(countRecords > 0);
@@ -109,10 +114,14 @@ default int recordOut() {
   }
 
   default void deleteBigQueryTable(String table) throws IOException, InterruptedException {
-    GcpClient.dropBqQuery(SeleniumHelper.readParameters(table));
+    BigQueryClient.dropBqQuery(SeleniumHelper.readParameters(table));
     BeforeActions.scenario.write("BigQuery Table Deleted Successfully");
   }
 
+  default void selectSourcePlugin(String pluginName) {
+    SeleniumDriver.getDriver().findElement(By.xpath("//*[@data-cy='plugin-" + pluginName + "-batchsource']")).click();
+  }
+
   default void selectSinkPlugin(String pluginName) {
     SeleniumDriver.getDriver().findElement(By.xpath("//*[@data-cy='plugin-" + pluginName + "-batchsink']")).click();
   }
@@ -127,4 +136,54 @@ default WebElement linkSinkPlugin(String pluginName) {
     String a = "//*[contains(@data-cy,'plugin-node-" + pluginName + "']";
     return SeleniumDriver.getDriver().findElement(By.xpath("//*[contains(@data-cy,'plugin-node-" + pluginName + "')]"));
   }
+
+  default void connectSourceAndSink(String source, String sink) {
+    By sinkNode = By.xpath("//*[contains(@data-cy,'plugin-node-" + sink + "') and @data-type='batchsink']");
+    SeleniumHelper.waitElementIsVisible(SeleniumDriver.getDriver().findElement(sinkNode));
+    SeleniumHelper.dragAndDrop(
+      SeleniumDriver.getDriver().findElement(By.xpath("//*[contains(@class,'plugin-endpoint_" + source + "')]")),
+      SeleniumDriver.getDriver().findElement(sinkNode));
+  }
+
+  default void openSourcePluginProperties(String plugin) {
+    SeleniumDriver.getDriver().findElement(
+      By.xpath("//*[contains(@data-cy,'plugin-node-" + plugin + "') and " +
+                 "@data-type='batchsource']//div[@class='node-metadata']/div[2]")).click();
+  }
+
+  default void openSinkPluginProperties(String plugin) {
+    SeleniumDriver.getDriver().findElement(
+      By.xpath("//*[contains(@data-cy,'plugin-node-" + plugin + "') and " +
+                 "@data-type='batchsink']//div[@class='node-metadata']/div[2]")).click();
+  }
+
+  default void openSourcePluginPreviewData(String plugin) {
+    SeleniumDriver.getDriver().findElement(
+      By.xpath("//*[@data-type='batchsource']//*[contains(@data-cy,'-preview-data-btn') " +
+                 "and contains(@data-cy,'" + plugin + "') and @class='node-preview-data-btn ng-scope']")).click();
+  }
+
+  default void openSinkPluginPreviewData(String plugin) {
+    SeleniumDriver.getDriver().findElement(
+      By.xpath("//*[@data-type='batchsink']//*[contains(@data-cy,'-preview-data-btn') " +
+                 "and contains(@data-cy,'" + plugin + "') and @class='node-preview-data-btn ng-scope']")).click();
+  }
+
+  default boolean compareTransferredRecords() {
+    int inCount = recordIn();
+    int outCount = recordOut();
+    return outCount != 0 && inCount == outCount;
+  }
+
+  default void validateSchema(Map<String, String> expectedOutputSchema) {
+    Map<String, String> actualOutputSchema = new HashMap<>();
+    int index = 0;
+    for (WebElement element : SCHEMA_LOCATORS.outputSchemaColumnNames) {
+      actualOutputSchema.put(element.getAttribute("value"),
+                             SCHEMA_LOCATORS.outputSchemaDataTypes.get(index).getAttribute("title"));
+      index++;
+    }
+    Assert.assertTrue("Schema displayed on UI should match with expected Schema",
+                      actualOutputSchema.equals(expectedOutputSchema));
+  }
 }
diff --git a/src/main/java/io/cdap/e2e/utils/ConstantsUtil.java b/src/main/java/io/cdap/e2e/utils/ConstantsUtil.java
index f4f3b65b2..7e384bccc 100644
--- a/src/main/java/io/cdap/e2e/utils/ConstantsUtil.java
+++ b/src/main/java/io/cdap/e2e/utils/ConstantsUtil.java
@@ -34,4 +34,8 @@ public class ConstantsUtil {
   public static final int ONE = 1;
   public static final int WAIT_TIME = 60;
   public static final String PASS_CASE = "First case passed";
+  public static final String ERROR_MSG_COLOR = "#a40403";
+  public static final String ERROR_MSG_MANDATORY = "Required property 'PROPERTY' has no value.";
+  public static final String DEFAULT_PLUGIN_PROPERTIES_FILE = "pluginParameters.properties";
+  public static final String DEFAULT_ERROR_PROPERTIES_FILE = "errorMessage.properties";
 }
diff --git a/src/main/java/io/cdap/e2e/utils/JsonUtils.java b/src/main/java/io/cdap/e2e/utils/JsonUtils.java
new file mode 100644
index 000000000..58d03c422
--- /dev/null
+++ b/src/main/java/io/cdap/e2e/utils/JsonUtils.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright © 2021 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.e2e.utils;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import com.google.gson.JsonSyntaxException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * JsonUtils contains the json helper functions.
+ */
+public class JsonUtils {
+  private static final Logger logger = LoggerFactory.getLogger(JsonUtils.class);
+
+  public static Map<String, String> convertKeyValueJsonArrayToMap(String json) {
+    Map<String, String> keyValueMap = new HashMap<>();
+    try {
+      JsonArray jsonArray = (JsonArray) new JsonParser().parse(json);
+      for (JsonElement jsonElement : jsonArray) {
+        keyValueMap.put(jsonElement.getAsJsonObject().get("key").getAsString(),
+                        jsonElement.getAsJsonObject().get("value").getAsString());
+      }
+    } catch (JsonSyntaxException e) {
+      logger.error("JsonUtils : Error while converting to Map - " + e);
+    } catch (Exception e) {
+      logger.error("JsonUtils : Error while converting to Map - key or value node not present in input json - " + e);
+    }
+    return keyValueMap;
+  }
+
+  public static int countJsonNodeSize(String json, String node) {
+    int size = 0;
+    try {
+      JsonObject jsonObject = (JsonObject) new JsonParser().parse(json);
+      if (jsonObject.get(node).isJsonArray()) {
+        JsonArray dataObject = jsonObject.getAsJsonArray(node);
+        size = dataObject.size();
+      } else if (!jsonObject.get(node).isJsonNull()) {
+        size = 1;
+      }
+    } catch (JsonSyntaxException e) {
+      logger.error("JsonUtils : Error while parsing JsonString - " + e);
+    } catch (NullPointerException e) {
+      logger.error("JsonUtils : Node \"" + node + "\" is not present in input json " + e);
+    }
+    return size;
+  }
+}
diff --git a/src/main/java/io/cdap/e2e/utils/PluginPropertyUtils.java b/src/main/java/io/cdap/e2e/utils/PluginPropertyUtils.java
new file mode 100644
index 000000000..db3c40ed0
--- /dev/null
+++ b/src/main/java/io/cdap/e2e/utils/PluginPropertyUtils.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright © 2021 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.e2e.utils;
+
+import org.junit.Assert;
+import org.openqa.selenium.By;
+import org.openqa.selenium.WebElement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+/**
+ * PluginPropertyUtils contains the e2e test helper functions.
+ */
+public class PluginPropertyUtils {
+
+  protected static Properties pluginProperties = new Properties();
+  private static final Properties errorProperties = new Properties();
+  private static final Logger logger = LoggerFactory.getLogger(PluginPropertyUtils.class);
+
+  static {
+    try {
+      pluginProperties.load(PluginPropertyUtils.class
+                              .getResourceAsStream("/" + ConstantsUtil.DEFAULT_PLUGIN_PROPERTIES_FILE));
+      errorProperties.load(PluginPropertyUtils.class
+                             .getResourceAsStream("/" + ConstantsUtil.DEFAULT_ERROR_PROPERTIES_FILE));
+    } catch (Exception e) {
+      logger.error("Error while reading properties file" + e);
+    }
+  }
+
+  public static String pluginProp(String property) {
+    return pluginProperties.getProperty(property);
+  }
+
+  public static String errorProp(String property) {
+    return errorProperties.getProperty(property);
+  }
+
+  public static void validateMandatoryPropertyError(String property) {
+    String expectedErrorMessage = ConstantsUtil.ERROR_MSG_MANDATORY.replaceAll("PROPERTY", property);
+    String actualErrorMessage = findPropertyErrorElement(property).getText();
+    Assert.assertEquals(expectedErrorMessage, actualErrorMessage);
+    String actualColor = PluginPropertyUtils.getErrorColor(PluginPropertyUtils.findPropertyErrorElement(property));
+    String expectedColor = ConstantsUtil.ERROR_MSG_COLOR;
+    Assert.assertEquals(expectedColor, actualColor);
+  }
+
+  public static WebElement findPropertyErrorElement(String property) {
+    return SeleniumDriver.getDriver().findElement(
+      By.xpath("//*[@data-cy='" + property + "']/following-sibling::div[@data-cy='property-row-error']"));
+  }
+
+  public static String getErrorColor(WebElement element) {
+    String color = element.getCssValue(ConstantsUtil.COLOR);
+    String[] hexValue = color.replace("rgba(", "").
+      replace(")", "").split(",");
+    int hexValue1 = Integer.parseInt(hexValue[0]);
+    hexValue[1] = hexValue[1].trim();
+    int hexValue2 = Integer.parseInt(hexValue[1]);
+    hexValue[2] = hexValue[2].trim();
+    int hexValue3 = Integer.parseInt(hexValue[2]);
+    return String.format("#%02x%02x%02x", hexValue1, hexValue2, hexValue3);
+  }
+}
diff --git a/src/main/java/io/cdap/e2e/utils/PropModifier.java b/src/main/java/io/cdap/e2e/utils/PropModifier.java
new file mode 100644
index 000000000..42ba37d53
--- /dev/null
+++ b/src/main/java/io/cdap/e2e/utils/PropModifier.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright © 2021 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.e2e.utils;
+
+import cucumber.api.event.EventHandler;
+import cucumber.api.event.EventListener;
+import cucumber.api.event.EventPublisher;
+import cucumber.api.event.TestRunFinished;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+/**
+ * Event listener to update properties on the run, called as a plugin by TestRunner.
+ */
+public class PropModifier implements EventListener {
+
+  private static final Logger logger = LoggerFactory.getLogger(PropModifier.class);
+
+  private static Properties pluginPropBackup;
+
+  public PropModifier(String fileName) {
+    appendToProps(fileName);
+  }
+
+  private void appendToProps(String fileName) {
+    try {
+      pluginPropBackup = (Properties) PluginPropertyUtils.pluginProperties.clone();
+      PluginPropertyUtils.pluginProperties.load(PropModifier.class.getResourceAsStream("/" + fileName));
+    } catch (Exception e) {
+      logger.error("Error while reading file " + fileName + ": " + e);
+    }
+  }
+
+  @Override
+  public void setEventPublisher(EventPublisher eventPublisher) {
+    //post action: reset properties to default
+    eventPublisher.registerHandlerFor(TestRunFinished.class, runFinishedHandler);
+  }
+
+  private EventHandler<TestRunFinished> runFinishedHandler = new EventHandler<TestRunFinished>() {
+    @Override
+    public void receive(TestRunFinished event) {
+      PluginPropertyUtils.pluginProperties = pluginPropBackup;
+    }
+  };
+}
diff --git a/src/main/java/io/cdap/e2e/utils/StorageClient.java b/src/main/java/io/cdap/e2e/utils/StorageClient.java
new file mode 100644
index 000000000..de65f74d0
--- /dev/null
+++ b/src/main/java/io/cdap/e2e/utils/StorageClient.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright © 2021 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.e2e.utils;
+
+import com.google.api.gax.paging.Page;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+/**
+ * Represents Google Cloud Storage client.
+ */
+public class StorageClient {
+
+  private static Storage storageService = null;
+
+  private static Storage getStorageService() throws IOException {
+    return (null == storageService) ? StorageOptions.newBuilder().setProjectId(
+      SeleniumHelper.readParameters(ConstantsUtil.PROJECT_ID)).build().getService() : storageService;
+  }
+
+  public static Page<Bucket> listBuckets() throws IOException {
+    return getStorageService().list();
+  }
+
+  public static Page<Blob> listObjects(String bucketName) throws IOException {
+    return getStorageService().list(bucketName);
+  }
+
+  public static Page<Blob> listObjectsWithPrefix(
+    String bucketName, String directoryPrefix) throws IOException {
+    return getStorageService().list(bucketName, Storage.BlobListOption.prefix(directoryPrefix),
+        Storage.BlobListOption.currentDirectory());
+  }
+
+  public static boolean deleteBucket(String bucketName) throws IOException {
+    return getStorageService().get(bucketName).delete();
+  }
+
+  public static boolean deleteObject(String bucketName, String objectName) throws IOException {
+    return getStorageService().delete(bucketName, objectName);
+  }
+
+  public static Blob getObjectMetadata(String bucketName, String blobName)
+    throws StorageException, IOException {
+    return getStorageService().get(bucketName, blobName, Storage.BlobGetOption.fields(Storage.BlobField.values()));
+  }
+
+  public static Bucket getBucketMetadata(String bucketName) throws IOException {
+    return getStorageService().get(bucketName, Storage.BucketGetOption.fields(Storage.BucketField.values()));
+  }
+
+  public static String getBucketCmekKey(String bucketName) throws IOException {
+    return getBucketMetadata(bucketName).getDefaultKmsKeyName();
+  }
+
+  public static String getObjectCmekKey(String bucketName, String blobName) throws IOException {
+    return getObjectMetadata(bucketName, blobName).getKmsKeyName();
+  }
+
+  public static Bucket createBucket(String bucketName) throws IOException {
+    return getStorageService().create(BucketInfo.of(bucketName));
+  }
+
+  public static Blob uploadObject(String bucketName, String objectName, String filePath)
+    throws IOException, URISyntaxException {
+    return getStorageService().create(
+      BlobInfo.newBuilder(BlobId.of(bucketName, objectName)).build(),
+      Files.readAllBytes(Paths.get(StorageClient.class.getResource("/" + filePath).toURI())));
+  }
+
+}
diff --git a/src/main/java/stepsdesign/BeforeActions.java b/src/main/java/stepsdesign/BeforeActions.java
index 815f60038..4b1b3bfeb 100644
--- a/src/main/java/stepsdesign/BeforeActions.java
+++ b/src/main/java/stepsdesign/BeforeActions.java
@@ -33,7 +33,7 @@ public class BeforeActions {
   public static Scenario scenario;
   public static File file;
 
-  @Before
+  @Before(order = 0)
   public void setUp(Scenario scenario) throws IOException {
     this.scenario = scenario;
     SeleniumDriver.setUpDriver();
diff --git a/src/main/resources/connectionParameters.properties b/src/main/resources/connectionParameters.properties
index 794b2e0e9..2b9861a38 100644
--- a/src/main/resources/connectionParameters.properties
+++ b/src/main/resources/connectionParameters.properties
@@ -1,7 +1,10 @@
-delimiter@=@
+# GCP project Id
 projectId=cdf-athena
+# Selenium browser window size
 windowSize=1920x1040
+# GCP datatest
 dataset=test_automation
+# CDAP application url
 cdfurl=http://localhost:11011/pipelines/ns/default/studio
 
 # properties to be removed-start
@@ -9,4 +12,5 @@ Project-ID=cdf-athena
 @TC-Demo-1_GCS=gs://cdf-athena-test/test.csv
 Data-Set=test_automation
 tableDemo=DemoCheck1
+delimiter@=@
 # properties to be removed-end