From 37d509bb4207aa5b381431714b3cb09ad498d9e6 Mon Sep 17 00:00:00 2001
From: albertshau <ashau@google.com>
Date: Wed, 13 Sep 2023 13:59:21 -0700
Subject: [PATCH] CDAP-20810 fix data cacher storage level

Fixed a bug where the data cacher would ignore the storage level
configured by the user.
---
 .../main/java/io/cdap/plugin/spark/DataCacher.java  | 13 ++-----------
 1 file changed, 2 insertions(+), 11 deletions(-)

diff --git a/spark-plugins/src/main/java/io/cdap/plugin/spark/DataCacher.java b/spark-plugins/src/main/java/io/cdap/plugin/spark/DataCacher.java
index 4a349561f..0e0e76408 100644
--- a/spark-plugins/src/main/java/io/cdap/plugin/spark/DataCacher.java
+++ b/spark-plugins/src/main/java/io/cdap/plugin/spark/DataCacher.java
@@ -58,17 +58,8 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
 
   @Override
   public JavaRDD<StructuredRecord> transform(SparkExecutionPluginContext sparkExecutionPluginContext,
-                                             JavaRDD<StructuredRecord> javaRDD) throws Exception {
-
-    StorageLevel storageLevel = StorageLevel.MEMORY_AND_DISK();
-    if (config.storageLevel.isEmpty()) {
-      StorageLevel.fromString(config.storageLevel);
-    }
-    if (storageLevel == StorageLevel.NONE()) {
-      throw new RuntimeException(
-        String.format("Invalid storage level '%s'. Please select a valid value", config.storageLevel));
-    }
-
+                                             JavaRDD<StructuredRecord> javaRDD) {
+    StorageLevel storageLevel = StorageLevel.fromString(config.storageLevel);
     javaRDD.persist(storageLevel);
     return javaRDD;
   }