SAMZA-1094 SAMZA-1101 SAMZA-1159; Remove MessageEnvelope from public operator APIs. : Delay the creation of SinkFunction for output streams. : Move StreamSpec from a public API to an internal class.

Removed the MessageEnvelope and OutputStream interfaces from public operator APIs.
Moved the creation of SinkFunction for output streams to SinkOperatorSpec.
Moved StreamSpec from a public API to an internal class.

Additionally,
1. Removed references to StreamGraph in OperatorSpecs. It was being used to getNextOpId(). MessageStreamsImpl now gets the ID and gives it to OperatorSpecs itself.
2. Updated and cleaned up the StreamGraphBuilder examples.
3. Renamed SinkOperatorSpec to OutputOperatorSpec since its used by sink, sendTo and partitionBy.

nickpan47 and xinyuiscool, please take a look.

Author: Prateek Maheshwari <[email protected]>

Reviewers: Jagadish <[email protected]>, Yi Pan <[email protected]>

Closes apache#92 from prateekm/message-envelope-removal

Author: prateekm

samza-api/src/main/java/org/apache/samza/operators/MessageStream.java

@@ -75,22 +75,24 @@ public interface MessageStream<M> {
   MessageStream<M> filter(FilterFunction<M> filterFn);
 
   /**
-   * Allows sending messages in this {@link MessageStream} to an output using the provided {@link SinkFunction}.
+   * Allows sending messages in this {@link MessageStream} to an output system using the provided {@link SinkFunction}.
    *
-   * NOTE: the output may not be a {@link org.apache.samza.system.SystemStream}. It can be an external database, etc.
+   * NOTE: If the output is for a {@link org.apache.samza.system.SystemStream}, use
+   * {@link #sendTo(OutputStream)} instead. This transform should only be used to output to
+   * non-stream systems (e.g., an external database).
    *
-   * @param sinkFn  the function to send messages in this stream to output
+   * @param sinkFn the function to send messages in this stream to an external system
    */
   void sink(SinkFunction<M> sinkFn);
 
   /**
    * Allows sending messages in this {@link MessageStream} to an output {@link MessageStream}.
    *
-   * NOTE: the {@code stream} has to be a {@link MessageStream}.
-   *
-   * @param stream  the output {@link MessageStream}
+   * @param outputStream the output stream to send messages to
+   * @param <K> the type of key in the outgoing message
+   * @param <V> the type of message in the outgoing message
    */
-  void sendTo(OutputStream<M> stream);
+  <K, V> void sendTo(OutputStream<K, V, M> outputStream);
 
   /**
    * Groups the messages in this {@link MessageStream} according to the provided {@link Window} semantics
@@ -128,19 +130,20 @@ public interface MessageStream<M> {
    * <p>
    * The merging streams must have the same messages of type {@code M}.
    *
-   * @param otherStreams  other {@link MessageStream}s to be merged with this {@link MessageStream}
-   * @return  the merged {@link MessageStream}
+   * @param otherStreams other {@link MessageStream}s to be merged with this {@link MessageStream}
+   * @return the merged {@link MessageStream}
    */
   MessageStream<M> merge(Collection<MessageStream<M>> otherStreams);
 
   /**
-   * Send the input message to an output {@link org.apache.samza.system.SystemStream} and consume it as input {@link MessageStream} again.
-   *
-   * Note: this is an transform function only used in logic DAG. In a physical DAG, this is either translated to a NOOP function, or a {@code MessageStream#sendThrough} function.
+   * Sends the messages of type {@code M}in this {@link MessageStream} to a repartitioned output stream and consumes
+   * them as an input {@link MessageStream} again. Uses keys returned by the {@code keyExtractor} as the partition key.
    *
-   * @param parKeyExtractor  a {@link Function} that extract the partition key from a message in this {@link MessageStream}
-   * @param <K>  the type of partition key
-   * @return  a {@link MessageStream} object after the re-partition
+   * @param keyExtractor the {@link Function} to extract the output message key and partition key from
+   *                     the input message
+   * @param <K> the type of output message key and partition key
+   * @return the repartitioned {@link MessageStream}
    */
-  <K> MessageStream<M> partitionBy(Function<M, K> parKeyExtractor);
+  <K> MessageStream<M> partitionBy(Function<M, K> keyExtractor);
+
 }

samza-api/src/main/java/org/apache/samza/operators/OutputStream.java

@@ -19,23 +19,15 @@
 package org.apache.samza.operators;
 
 import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.functions.SinkFunction;
-
 
 /**
- * The interface class defining the specific {@link SinkFunction} for a system {@link OutputStream}.
+ * An output stream to send messages to.
  *
- * @param <M>  The type of message to be send to this output stream
+ * @param <K> the type of key in the outgoing message
+ * @param <V> the type of message in the outgoing message
+ * @param <M> the type of message in this {@link OutputStream}
  */
 @InterfaceStability.Unstable
-public interface OutputStream<M> {
+public interface OutputStream<K, V, M> {
 
-  /**
-   * Returns the specific {@link SinkFunction} for this output stream. The {@link OutputStream} is created
-   * via {@link StreamGraph#createOutStream(StreamSpec, Serde, Serde)} or {@link StreamGraph#createIntStream(StreamSpec, Serde, Serde)}.
-   * Hence, the proper types of serdes for key and value are instantiated and are used in the {@link SinkFunction} returned.
-   *
-   * @return  The pre-defined {@link SinkFunction} to apply proper serdes before sending the message to the output stream.
-   */
-  SinkFunction<M> getSinkFunction();
 }

samza-api/src/main/java/org/apache/samza/operators/StreamGraph.java

@@ -19,76 +19,52 @@
 package org.apache.samza.operators;
 
 import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.data.MessageEnvelope;
-import org.apache.samza.serializers.Serde;
-import org.apache.samza.system.StreamSpec;
 
-import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 
 /**
- * Job-level programming interface to create an operator DAG and run in various different runtime environments.
+ * Provides APIs for accessing {@link MessageStream}s to be used to create the DAG of transforms.
  */
 @InterfaceStability.Unstable
 public interface StreamGraph {
-  /**
-   * Method to add an input {@link MessageStream} from the system
-   *
-   * @param streamSpec  the {@link StreamSpec} describing the physical characteristics of the input {@link MessageStream}
-   * @param keySerde  the serde used to serialize/deserialize the message key from the input {@link MessageStream}
-   * @param msgSerde  the serde used to serialize/deserialize the message body from the input {@link MessageStream}
-   * @param <K>  the type of key in the input message
-   * @param <V>  the type of message in the input message
-   * @param <M>  the type of {@link MessageEnvelope} in the input {@link MessageStream}
-   * @return   the input {@link MessageStream} object
-   */
-  <K, V, M extends MessageEnvelope<K, V>> MessageStream<M> createInStream(StreamSpec streamSpec, Serde<K> keySerde, Serde<V> msgSerde);
-
-  /**
-   * Method to add an output {@link MessageStream} from the system
-   *
-   * @param streamSpec  the {@link StreamSpec} describing the physical characteristics of the output {@link MessageStream}
-   * @param keySerde  the serde used to serialize/deserialize the message key from the output {@link MessageStream}
-   * @param msgSerde  the serde used to serialize/deserialize the message body from the output {@link MessageStream}
-   * @param <K>  the type of key in the output message
-   * @param <V>  the type of message in the output message
-   * @param <M>  the type of {@link MessageEnvelope} in the output {@link MessageStream}
-   * @return   the output {@link MessageStream} object
-   */
-  <K, V, M extends MessageEnvelope<K, V>> OutputStream<M> createOutStream(StreamSpec streamSpec, Serde<K> keySerde, Serde<V> msgSerde);
-
-  /**
-   * Method to add an intermediate {@link MessageStream} from the system
-   *
-   * @param streamSpec  the {@link StreamSpec} describing the physical characteristics of the intermediate {@link MessageStream}
-   * @param keySerde  the serde used to serialize/deserialize the message key from the intermediate {@link MessageStream}
-   * @param msgSerde  the serde used to serialize/deserialize the message body from the intermediate {@link MessageStream}
-   * @param <K>  the type of key in the intermediate message
-   * @param <V>  the type of message in the intermediate message
-   * @param <M>  the type of {@link MessageEnvelope} in the intermediate {@link MessageStream}
-   * @return   the intermediate {@link MessageStream} object
-   */
-  <K, V, M extends MessageEnvelope<K, V>> OutputStream<M> createIntStream(StreamSpec streamSpec, Serde<K> keySerde, Serde<V> msgSerde);
 
   /**
-   * Method to get the input {@link MessageStream}s
+   * Gets the input {@link MessageStream} corresponding to the logical {@code streamId}.
    *
+   * @param streamId the unique logical ID for the stream
+   * @param msgBuilder the {@link BiFunction} to convert the incoming key and message to a message
+   *                   in the input {@link MessageStream}
+   * @param <K> the type of key in the incoming message
+   * @param <V> the type of message in the incoming message
+   * @param <M> the type of message in the input {@link MessageStream}
    * @return the input {@link MessageStream}
    */
-  Map<StreamSpec, MessageStream> getInStreams();
+  <K, V, M> MessageStream<M> getInputStream(String streamId, BiFunction<K, V, M> msgBuilder);
 
   /**
-   * Method to get the {@link OutputStream}s
+   * Gets the {@link OutputStream} corresponding to the logical {@code streamId}.
    *
-   * @return  the map of all {@link OutputStream}s
+   * @param streamId the unique logical ID for the stream
+   * @param keyExtractor the {@link Function} to extract the outgoing key from the output message
+   * @param msgExtractor the {@link Function} to extract the outgoing message from the output message
+   * @param <K> the type of key in the outgoing message
+   * @param <V> the type of message in the outgoing message
+   * @param <M> the type of message in the {@link OutputStream}
+   * @return the output {@link MessageStream}
    */
-  Map<StreamSpec, OutputStream> getOutStreams();
+  <K, V, M> OutputStream<K, V, M> getOutputStream(String streamId,
+      Function<M, K> keyExtractor, Function<M, V> msgExtractor);
 
   /**
-   * Method to set the {@link ContextManager} for this {@link StreamGraph}
+   * Sets the {@link ContextManager} for this {@link StreamGraph}.
+   *
+   * The provided {@code contextManager} will be initialized before the transformation functions
+   * and can be used to setup shared context between them.
    *
-   * @param manager  the {@link ContextManager} object
-   * @return  this {@link StreamGraph} object
+   * @param contextManager the {@link ContextManager} to use for the {@link StreamGraph}
+   * @return the {@link StreamGraph} with the {@code contextManager} as its {@link ContextManager}
    */
-  StreamGraph withContextManager(ContextManager manager);
+  StreamGraph withContextManager(ContextManager contextManager);
 
 }

samza-api/src/main/java/org/apache/samza/operators/data/InputMessageEnvelope.java

@@ -37,7 +37,7 @@ public interface SinkFunction<M>  extends InitableFunction {
    * or shut the container down.
    *
    * @param message  the input message to be sent to an output {@link org.apache.samza.system.SystemStream}
-   * @param messageCollector  the {@link MessageCollector} to send the {@link org.apache.samza.operators.data.MessageEnvelope}
+   * @param messageCollector  the {@link MessageCollector} to send the message
    * @param taskCoordinator  the {@link TaskCoordinator} to request commits or shutdown
    */
   void apply(M message, MessageCollector messageCollector, TaskCoordinator taskCoordinator);

samza-api/src/main/java/org/apache/samza/operators/data/LongOffset.java

@@ -18,14 +18,15 @@
  */
 package org.apache.samza.runtime;
 
-import java.lang.reflect.Constructor;
 import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.ConfigException;
 import org.apache.samza.job.ApplicationStatus;
 import org.apache.samza.system.StreamSpec;
 
+import java.lang.reflect.Constructor;
+
 
 /**
  * A physical execution engine to deploy the config and jobs to run the {@link org.apache.samza.operators.StreamGraph}