Simplified Streamlet interface (#2519)

* Removed KVStreamlet

* Fix build

* Fixed checkstyle

* Fix bug in JoinTest

* Fixed jointest

examples/src/java/com/twitter/heron/examples/streamlet/WordCountStreamletTopology.java

@@ -18,7 +18,6 @@
 
 import com.twitter.heron.streamlet.Builder;
 import com.twitter.heron.streamlet.Config;
-import com.twitter.heron.streamlet.KeyValue;
 import com.twitter.heron.streamlet.Runner;
 import com.twitter.heron.streamlet.WindowConfig;
 
@@ -51,8 +50,7 @@ public static void main(String[] args) {
     Builder builder = Builder.createBuilder();
     builder.newSource(() -> "Mary had a little lamb")
         .flatMap((sentence) -> Arrays.asList(sentence.split("\\s+")))
-        .mapToKV((word) -> new KeyValue<>(word, 1))
-        .reduceByKeyAndWindow(WindowConfig.TumblingCountWindow(10), (x, y) -> x + y)
+        .reduceByKeyAndWindow(x -> x, WindowConfig.TumblingCountWindow(10), 0, (x, y) -> x + 1)
         .log();
     Config conf = new Config();
     conf.setNumContainers(parallelism);

heron/api/src/java/com/twitter/heron/streamlet/Builder.java

@@ -32,29 +32,11 @@ static Builder createBuilder() {
    */
   <R> Streamlet<R> newSource(SerializableSupplier<R> supplier);
 
-  /**
-   * Same as above except returns KVStreamlet
-   * @param supplier
-   * @param <K>
-   * @param <V>
-   * @return
-   */
-  <K, V> KVStreamlet<K, V> newKVSource(SerializableSupplier<KeyValue<K, V>> supplier);
-
   /**
    * Creates a new Streamlet using the underlying generator
    * @param generator The generator that generates the tuples of the streamlet
    * @param <R>
    * @return
    */
   <R> Streamlet<R> newSource(Source<R> generator);
-
-  /**
-   * Same as above except returns a KVStreamlet
-   * @param generator The genertor that generates the tuples of the streamlet
-   * @param <K>
-   * @param <V>
-   * @return
-   */
-  <K, V> KVStreamlet<K, V> newKVSource(Source<KeyValue<K, V>> generator);
 }