[HUDI-652] Decouple HoodieReadClient and AbstractHoodieClient to break the inheritance chain (#1372)

* Removed timeline server support
* Removed try-with-resource

hudi-client/src/main/java/org/apache/hudi/client/HoodieReadClient.java

@@ -25,7 +25,6 @@ import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.table.HoodieTimeline;
 import org.apache.hudi.common.util.CompactionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.collection.Pair;
@@ -46,6 +45,7 @@ import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.sql.types.StructType;
 
+import java.io.Serializable;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -56,7 +56,7 @@ import scala.Tuple2;
 /**
  * Provides an RDD based API for accessing/filtering Hoodie tables, based on keys.
  */
-public class HoodieReadClient<T extends HoodieRecordPayload> extends AbstractHoodieClient {
+public class HoodieReadClient<T extends HoodieRecordPayload> implements Serializable {
 
   private static final Logger LOG = LogManager.getLogger(HoodieReadClient.class);
 
@@ -65,9 +65,9 @@ public class HoodieReadClient<T extends HoodieRecordPayload> extends AbstractHoo
    * basepath pointing to the table. Until, then just always assume a BloomIndex
    */
   private final transient HoodieIndex<T> index;
-  private final HoodieTimeline commitTimeline;
   private HoodieTable hoodieTable;
   private transient Option<SQLContext> sqlContextOpt;
+  private final transient JavaSparkContext jsc;
 
   /**
    * @param basePath path to Hoodie table
@@ -108,12 +108,11 @@ public class HoodieReadClient<T extends HoodieRecordPayload> extends AbstractHoo
    */
   public HoodieReadClient(JavaSparkContext jsc, HoodieWriteConfig clientConfig,
       Option<EmbeddedTimelineService> timelineService) {
-    super(jsc, clientConfig, timelineService);
+    this.jsc = jsc;
     final String basePath = clientConfig.getBasePath();
     // Create a Hoodie table which encapsulated the commits and files visible
     HoodieTableMetaClient metaClient = new HoodieTableMetaClient(jsc.hadoopConfiguration(), basePath, true);
     this.hoodieTable = HoodieTable.getHoodieTable(metaClient, clientConfig, jsc);
-    this.commitTimeline = metaClient.getCommitTimeline().filterCompletedInstants();
     this.index = HoodieIndex.createIndex(clientConfig, jsc);
     this.sqlContextOpt = Option.empty();
   }

hudi-client/src/test/java/org/apache/hudi/client/TestHoodieReadClient.java

@@ -96,8 +96,8 @@ public class TestHoodieReadClient extends TestHoodieClientBase {
    */
   private void testReadFilterExist(HoodieWriteConfig config,
       Function3<JavaRDD<WriteStatus>, HoodieWriteClient, JavaRDD<HoodieRecord>, String> writeFn) throws Exception {
-    try (HoodieWriteClient writeClient = getHoodieWriteClient(config);
+    try (HoodieWriteClient writeClient = getHoodieWriteClient(config);) {
-        HoodieReadClient readClient = getHoodieReadClient(config.getBasePath());) {
+      HoodieReadClient readClient = getHoodieReadClient(config.getBasePath());
       String newCommitTime = writeClient.startCommit();
       List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 100);
       JavaRDD<HoodieRecord> recordsRDD = jsc.parallelize(records, 1);
@@ -113,7 +113,7 @@ public class TestHoodieReadClient extends TestHoodieClientBase {
       // Verify there are no errors
       assertNoWriteErrors(statuses);
 
-      try (HoodieReadClient anotherReadClient = getHoodieReadClient(config.getBasePath());) {
+      HoodieReadClient anotherReadClient = getHoodieReadClient(config.getBasePath());
       filteredRDD = anotherReadClient.filterExists(recordsRDD);
       List<HoodieRecord> result = filteredRDD.collect();
       // Check results
@@ -145,7 +145,6 @@ public class TestHoodieReadClient extends TestHoodieClientBase {
       assertEquals(0, anotherReadClient.getPendingCompactions().size());
     }
   }
-  }
 
   /**
    * Test tagLocation API after insert().

hudi-client/src/test/java/org/apache/hudi/table/TestMergeOnReadTable.java

@@ -759,7 +759,8 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
 
       List<HoodieRecord> updatedRecords = dataGen.generateUpdates(newCommitTime, records);
       JavaRDD<HoodieRecord> updatedRecordsRDD = jsc.parallelize(updatedRecords, 1);
-      try (HoodieReadClient readClient = new HoodieReadClient(jsc, config);) {
+
+      HoodieReadClient readClient = new HoodieReadClient(jsc, config);
       updatedRecords = readClient.tagLocation(updatedRecordsRDD).collect();
 
       // Write them to corresponding avro logfiles
@@ -809,7 +810,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
       }
     }
   }
-  }
 
   @Test
   public void testSimpleInsertsGeneratedIntoLogFiles() throws Exception {

hudi-client/src/test/java/org/apache/hudi/table/compact/TestAsyncCompaction.java

@@ -92,9 +92,8 @@ public class TestAsyncCompaction extends TestHoodieClientBase {
   public void testRollbackForInflightCompaction() throws Exception {
     // Rollback inflight compaction
     HoodieWriteConfig cfg = getConfig(false);
-    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);
+    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);) {
-         HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());) {
+      HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());
-
       String firstInstantTime = "001";
       String secondInstantTime = "004";
       String compactionInstantTime = "005";
@@ -155,9 +154,8 @@ public class TestAsyncCompaction extends TestHoodieClientBase {
 
     int numRecs = 2000;
 
-    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);
+    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);) {
-         HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());) {
+      HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());
-
       List<HoodieRecord> records = dataGen.generateInserts(firstInstantTime, numRecs);
       records = runNextDeltaCommits(client, readClient, Arrays.asList(firstInstantTime, secondInstantTime), records, cfg, true,
           new ArrayList<>());
@@ -197,9 +195,8 @@ public class TestAsyncCompaction extends TestHoodieClientBase {
   public void testInflightCompaction() throws Exception {
     // There is inflight compaction. Subsequent compaction run must work correctly
     HoodieWriteConfig cfg = getConfig(true);
-    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);
+    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);) {
-         HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());) {
+      HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());
-
       String firstInstantTime = "001";
       String secondInstantTime = "004";
       String compactionInstantTime = "005";
@@ -351,9 +348,8 @@ public class TestAsyncCompaction extends TestHoodieClientBase {
   public void testCompactionAfterTwoDeltaCommits() throws Exception {
     // No Delta Commits after compaction request
     HoodieWriteConfig cfg = getConfig(true);
-    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);
+    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);) {
-         HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());) {
+      HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());
-
       String firstInstantTime = "001";
       String secondInstantTime = "004";
       String compactionInstantTime = "005";
@@ -373,9 +369,8 @@ public class TestAsyncCompaction extends TestHoodieClientBase {
   public void testInterleavedCompaction() throws Exception {
     // Case: Two delta commits before and after compaction schedule
     HoodieWriteConfig cfg = getConfig(true);
-    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);
+    try (HoodieWriteClient client = getHoodieWriteClient(cfg, true);) {
-         HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());) {
+      HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath());
-
       String firstInstantTime = "001";
       String secondInstantTime = "004";
       String compactionInstantTime = "005";

hudi-spark/src/main/java/org/apache/hudi/DataSourceUtils.java

@@ -223,7 +223,8 @@ public class DataSourceUtils {
   @SuppressWarnings("unchecked")
   public static JavaRDD<HoodieRecord> dropDuplicates(JavaSparkContext jssc, JavaRDD<HoodieRecord> incomingHoodieRecords,
                                                      HoodieWriteConfig writeConfig, Option<EmbeddedTimelineService> timelineService) {
-    try (HoodieReadClient client = new HoodieReadClient<>(jssc, writeConfig, timelineService)) {
+    try {
+      HoodieReadClient client = new HoodieReadClient<>(jssc, writeConfig, timelineService);
       return client.tagLocation(incomingHoodieRecords)
           .filter(r -> !((HoodieRecord<HoodieRecordPayload>) r).isCurrentLocationKnown());
     } catch (TableNotFoundException e) {