SHS-NG M1: Remove unused methods from KVStore.

Turns out I ended up not using the raw storage methods in KVStore, so
this change removes them to simplify the API and save some code.

common/kvstore/src/main/java/org/apache/spark/kvstore/KVStore.java

@@ -25,21 +25,21 @@
  * Abstraction for a local key/value store for storing app data.
  *
  * <p>
- * Use {@link KVStoreBuilder} to create an instance. There are two main features provided by the
- * implementations of this interface:
+ * There are two main features provided by the implementations of this interface:
  * </p>
  *
- * <ul>
- *   <li>serialization: this feature is not optional; data will be serialized to and deserialized
- *   from the underlying data store using a {@link KVStoreSerializer}, which can be customized by
- *   the application. The serializer is based on Jackson, so it supports all the Jackson annotations
- *   for controlling the serialization of app-defined types.</li>
+ * <h3>Serialization</h3>
  *
- *   <li>key management: by using {@link #read(Class, Object)} and {@link #write(Class, Object)},
- *   applications can leave key management to the implementation. For applications that want to
- *   manage their own keys, the {@link #get(byte[], Class)} and {@link #set(byte[], Object)} methods
- *   are available.</li>
- * </ul>
+ * <p>
+ * Data will be serialized to and deserialized from the underlying data store using a
+ * {@link KVStoreSerializer}, which can be customized by the application. The serializer is
+ * based on Jackson, so it supports all the Jackson annotations for controlling the serialization
+ * of app-defined types.
+ * </p>
+ *
+ * <p>
+ * Data is also automatically compressed to save disk space.
+ * </p>
  *
  * <h3>Automatic Key Management</h3>
  *
@@ -78,26 +78,6 @@ public interface KVStore extends Closeable {
    */
   void setMetadata(Object value) throws Exception;
 
-  /**
-   * Returns the value of a specific key, deserialized to the given type.
-   */
-  <T> T get(byte[] key, Class<T> klass) throws Exception;
-
-  /**
-   * Write a single key directly to the store, atomically.
-   */
-  void put(byte[] key, Object value) throws Exception;
-
-  /**
-   * Removes a key from the store.
-   */
-  void delete(byte[] key) throws Exception;
-
-  /**
-   * Returns an iterator that will only list values with keys starting with the given prefix.
-   */
-  <T> KVStoreIterator<T> iterator(byte[] prefix, Class<T> klass) throws Exception;
-
   /**
    * Read a specific instance of an object.
    */

common/kvstore/src/main/java/org/apache/spark/kvstore/LevelDB.java

@@ -109,31 +109,19 @@ public void setMetadata(Object value) throws Exception {
     }
   }
 
-  @Override
-  public <T> T get(byte[] key, Class<T> klass) throws Exception {
+  <T> T get(byte[] key, Class<T> klass) throws Exception {
     byte[] data = db().get(key);
     if (data == null) {
       throw new NoSuchElementException(new String(key, UTF_8));
     }
     return serializer.deserialize(data, klass);
   }
 
-  @Override
-  public void put(byte[] key, Object value) throws Exception {
+  private void put(byte[] key, Object value) throws Exception {
     Preconditions.checkArgument(value != null, "Null values are not allowed.");
     db().put(key, serializer.serialize(value));
   }
 
-  @Override
-  public void delete(byte[] key) throws Exception {
-    db().delete(key);
-  }
-
-  @Override
-  public <T> KVStoreIterator<T> iterator(byte[] prefix, Class<T> klass) throws Exception {
-    throw new UnsupportedOperationException();
-  }
-
   @Override
   public <T> T read(Class<T> klass, Object naturalKey) throws Exception {
     Preconditions.checkArgument(naturalKey != null, "Null keys are not allowed.");

common/kvstore/src/test/java/org/apache/spark/kvstore/LevelDBSuite.java

@@ -74,29 +74,6 @@ public void testReopenAndVersionCheckDb() throws Exception {
     }
   }
 
-  @Test
-  public void testStringWriteReadDelete() throws Exception {
-    String string = "testString";
-    byte[] key = string.getBytes(UTF_8);
-    testReadWriteDelete(key, string);
-  }
-
-  @Test
-  public void testIntWriteReadDelete() throws Exception {
-    int value = 42;
-    byte[] key = "key".getBytes(UTF_8);
-    testReadWriteDelete(key, value);
-  }
-
-  @Test
-  public void testSimpleTypeWriteReadDelete() throws Exception {
-    byte[] key = "testKey".getBytes(UTF_8);
-    CustomType1 t = new CustomType1();
-    t.id = "id";
-    t.name = "name";
-    testReadWriteDelete(key, t);
-  }
-
   @Test
   public void testObjectWriteReadDelete() throws Exception {
     CustomType1 t = new CustomType1();
@@ -268,26 +245,6 @@ private int countKeys(Class<?> type) throws Exception {
     return count;
   }
 
-  private <T> void testReadWriteDelete(byte[] key, T value) throws Exception {
-    try {
-      db.get(key, value.getClass());
-      fail("Expected exception for non-existent key.");
-    } catch (NoSuchElementException nsee) {
-      // Expected.
-    }
-
-    db.put(key, value);
-    assertEquals(value, db.get(key, value.getClass()));
-
-    db.delete(key);
-    try {
-      db.get(key, value.getClass());
-      fail("Expected exception for deleted key.");
-    } catch (NoSuchElementException nsee) {
-      // Expected.
-    }
-  }
-
   public static class IntKeyType {
 
     @KVIndex