You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ni...@apache.org on 2016/12/01 22:51:06 UTC
[6/6] samza git commit: SAMZA-1054: Refactor Operator APIs
SAMZA-1054: Refactor Operator APIs
Some suggestions for an Operator API refactor and misc. cleanup. It does contain some implementation changes, mostly due to deleted, extracted or merged classes. (e.g. OperatorFactory + ChainedOperators == OperatorImpls).
Since git marked several moved classes as (delete + new) instead, it's probably best to apply the diff locally and browse the code in an IDE.
Some of the changes, in no particular order:
* Extracted XFunction interfaces into a .functions package in -api.
* -api's internal.Operators is now the -operators's spec.* package. Extracted interfaces and classes. Factory methods are now in OperatorSpecs.
* -api's MessageStreams is now -api's MessageStream interface and -operators's MessageStreamImpl.
* -api's internal.Windows classes are now in -api's .window package. Extracted interfaces and classes, but no implementation changes.
* OperatorFactory + ChainedOperators is now OperatorImpls, which is used from StreamOperatorAdaptorTask.
* Added a NoOpOperatorImpl, which acts as the root node for the OperatorImpl DAG returned by OperatorImpls.
* Removed usages of reactivestreams APIs since current code looks simpler without them. We can add them back when we need features like backpressure etc.
* Removed the InputSystemMessage interface.
* Made field names consistent (e.g Fn suffix for functions everywhere etc.).
* Some method/class visibility changes due to moved classes.
* General documentation changes, mostly to make public APIs clearer.
There are additional questions/tasks that we can address in future RBs:
* Updating Window and Trigger APIs.
* Merging samza-operator into samza-core.
* Questions about Message timestamp and Offset comparison semantics.
* Questions about OperatorSpec serialization (e.g. ID generation).
* Questions about StateStoreImpl and StoreFunctions.
Author: Prateek Maheshwari <pm...@linkedin.com>
Reviewers: Yi Pan <ni...@gmail.com>, Jagadish <ja...@gmail.com>
Closes #25 from prateekm/master
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/00543804
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/00543804
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/00543804
Branch: refs/heads/master
Commit: 00543804b3c32f1cbea0212e4a94e360b5a324cc
Parents: a980c96
Author: Prateek Maheshwari <pm...@linkedin.com>
Authored: Thu Dec 1 14:50:52 2016 -0800
Committer: Yi Pan (Data Infrastructure) <ni...@gmail.com>
Committed: Thu Dec 1 14:50:52 2016 -0800
----------------------------------------------------------------------
.../apache/samza/operators/MessageStream.java | 197 +++-----
.../apache/samza/operators/MessageStreams.java | 81 ----
.../samza/operators/StreamOperatorTask.java | 51 ++
.../apache/samza/operators/TriggerBuilder.java | 323 -------------
.../org/apache/samza/operators/WindowState.java | 81 ----
.../org/apache/samza/operators/Windows.java | 203 --------
.../operators/data/IncomingSystemMessage.java | 76 ---
.../data/IncomingSystemMessageEnvelope.java | 63 +++
.../operators/data/InputSystemMessage.java | 45 --
.../apache/samza/operators/data/Message.java | 64 ---
.../samza/operators/data/MessageEnvelope.java | 54 +++
.../operators/functions/FilterFunction.java | 40 ++
.../operators/functions/FlatMapFunction.java | 44 ++
.../samza/operators/functions/JoinFunction.java | 44 ++
.../samza/operators/functions/MapFunction.java | 41 ++
.../samza/operators/functions/SinkFunction.java | 46 ++
.../samza/operators/internal/Operators.java | 469 -------------------
.../samza/operators/internal/Trigger.java | 95 ----
.../samza/operators/internal/WindowFn.java | 60 ---
.../samza/operators/internal/WindowOutput.java | 55 ---
.../operators/task/StreamOperatorTask.java | 45 --
.../samza/operators/windows/SessionWindow.java | 102 ++++
.../samza/operators/windows/StoreFunctions.java | 67 +++
.../apache/samza/operators/windows/Trigger.java | 94 ++++
.../samza/operators/windows/TriggerBuilder.java | 320 +++++++++++++
.../apache/samza/operators/windows/Window.java | 49 ++
.../samza/operators/windows/WindowFn.java | 59 +++
.../samza/operators/windows/WindowOutput.java | 51 ++
.../samza/operators/windows/WindowState.java | 85 ++++
.../apache/samza/operators/windows/Windows.java | 100 ++++
.../org/apache/samza/operators/TestMessage.java | 47 --
.../samza/operators/TestMessageEnvelope.java | 61 +++
.../samza/operators/TestMessageStream.java | 180 -------
.../samza/operators/TestMessageStreams.java | 35 --
.../samza/operators/TestOutputMessage.java | 47 --
.../operators/TestOutputMessageEnvelope.java | 43 ++
.../samza/operators/TestTriggerBuilder.java | 214 ---------
.../org/apache/samza/operators/TestWindows.java | 106 -----
.../data/TestIncomingSystemMessage.java | 5 +-
.../samza/operators/data/TestLongOffset.java | 9 +-
.../samza/operators/internal/TestOperators.java | 128 -----
.../samza/operators/internal/TestTrigger.java | 68 ---
.../operators/internal/TestWindowOutput.java | 36 --
.../samza/operators/windows/TestTrigger.java | 68 +++
.../operators/windows/TestTriggerBuilder.java | 226 +++++++++
.../operators/windows/TestWindowOutput.java | 36 ++
.../samza/operators/windows/TestWindows.java | 109 +++++
.../samza/operators/MessageStreamImpl.java | 134 ++++++
.../apache/samza/operators/StateStoreImpl.java | 56 +++
.../operators/StreamOperatorAdaptorTask.java | 105 +++++
.../samza/operators/impl/ChainedOperators.java | 119 -----
.../samza/operators/impl/OperatorFactory.java | 85 ----
.../samza/operators/impl/OperatorImpl.java | 76 +--
.../samza/operators/impl/OperatorImpls.java | 125 +++++
.../operators/impl/PartialJoinOperatorImpl.java | 46 ++
.../samza/operators/impl/ProcessorContext.java | 53 ---
.../samza/operators/impl/RootOperatorImpl.java | 36 ++
.../impl/SessionWindowOperatorImpl.java | 67 +++
.../operators/impl/SimpleOperatorImpl.java | 49 --
.../samza/operators/impl/SinkOperatorImpl.java | 22 +-
.../samza/operators/impl/StateStoreImpl.java | 56 ---
.../operators/impl/StreamOperatorImpl.java | 47 ++
.../operators/impl/join/PartialJoinOpImpl.java | 44 --
.../impl/window/SessionWindowImpl.java | 65 ---
.../samza/operators/spec/OperatorSpec.java | 37 ++
.../samza/operators/spec/OperatorSpecs.java | 116 +++++
.../operators/spec/PartialJoinOperatorSpec.java | 104 ++++
.../samza/operators/spec/SinkOperatorSpec.java | 62 +++
.../operators/spec/StreamOperatorSpec.java | 67 +++
.../operators/spec/WindowOperatorSpec.java | 119 +++++
.../samza/task/StreamOperatorAdaptorTask.java | 85 ----
.../apache/samza/operators/BroadcastTask.java | 101 ++++
.../org/apache/samza/operators/JoinTask.java | 77 +++
.../operators/TestFluentStreamAdaptorTask.java | 85 ++++
.../samza/operators/TestFluentStreamTasks.java | 112 +++++
.../samza/operators/TestMessageStreamImpl.java | 203 ++++++++
.../samza/operators/TestStateStoreImpl.java | 72 +++
.../org/apache/samza/operators/WindowTask.java | 70 +++
.../data/JsonIncomingSystemMessageEnvelope.java | 60 +++
.../operators/impl/TestChainedOperators.java | 129 -----
.../operators/impl/TestOperatorFactory.java | 93 ----
.../samza/operators/impl/TestOperatorImpl.java | 48 +-
.../samza/operators/impl/TestOperatorImpls.java | 183 ++++++++
.../operators/impl/TestProcessorContext.java | 40 --
.../operators/impl/TestSessionWindowImpl.java | 111 +++++
.../operators/impl/TestSimpleOperatorImpl.java | 55 ---
.../operators/impl/TestSinkOperatorImpl.java | 25 +-
.../operators/impl/TestStateStoreImpl.java | 69 ---
.../operators/impl/TestStreamOperatorImpl.java | 60 +++
.../impl/window/TestSessionWindowImpl.java | 105 -----
.../samza/operators/spec/TestOperatorSpecs.java | 114 +++++
.../samza/task/BroadcastOperatorTask.java | 102 ----
.../samza/task/InputJsonSystemMessage.java | 67 ---
.../org/apache/samza/task/JoinOperatorTask.java | 80 ----
.../task/TestStreamOperatorAdaptorTask.java | 80 ----
.../samza/task/TestStreamOperatorTasks.java | 105 -----
.../apache/samza/task/WindowOperatorTask.java | 71 ---
97 files changed, 4374 insertions(+), 4240 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/MessageStream.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/MessageStream.java b/samza-api/src/main/java/org/apache/samza/operators/MessageStream.java
index dede631..d18536b 100644
--- a/samza-api/src/main/java/org/apache/samza/operators/MessageStream.java
+++ b/samza-api/src/main/java/org/apache/samza/operators/MessageStream.java
@@ -16,173 +16,112 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.samza.operators;
import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.data.Message;
-import org.apache.samza.operators.internal.Operators;
-import org.apache.samza.operators.internal.Operators.Operator;
-import org.apache.samza.operators.internal.WindowOutput;
-import org.apache.samza.task.MessageCollector;
-import org.apache.samza.task.TaskCoordinator;
+import org.apache.samza.operators.data.MessageEnvelope;
+import org.apache.samza.operators.functions.FilterFunction;
+import org.apache.samza.operators.functions.FlatMapFunction;
+import org.apache.samza.operators.functions.JoinFunction;
+import org.apache.samza.operators.functions.MapFunction;
+import org.apache.samza.operators.functions.SinkFunction;
+import org.apache.samza.operators.windows.Window;
+import org.apache.samza.operators.windows.WindowOutput;
+import org.apache.samza.operators.windows.WindowState;
-import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.function.BiFunction;
-import java.util.function.Function;
/**
- * This class defines either the input or output streams to/from the operators. Users use the API methods defined here to
- * directly program the stream processing stages that processes a stream and generate another one.
+ * Represents a stream of {@link MessageEnvelope}s.
+ * <p>
+ * A {@link MessageStream} can be transformed into another {@link MessageStream} by applying the transforms in this API.
*
- * @param <M> Type of message in this stream
+ * @param <M> type of {@link MessageEnvelope}s in this stream
*/
@InterfaceStability.Unstable
-public class MessageStream<M extends Message> {
-
- private final Set<Operator> subscribers = new HashSet<>();
-
- /**
- * Helper method to get the corresponding list of subscribers to a specific {@link MessageStream}.
- *
- * NOTE: This is purely an internal API and should not be used directly by programmers.
- *
- * @return A unmodifiable set containing all {@link Operator}s that subscribe to this {@link MessageStream} object
- */
- public Collection<Operator> getSubscribers() {
- return Collections.unmodifiableSet(this.subscribers);
- }
-
- /**
- * Public API methods start here
- */
+public interface MessageStream<M extends MessageEnvelope> {
/**
- * Defines a function API that takes three input parameters w/ types {@code A}, {@code B}, and {@code C} and w/o a return value
+ * Applies the provided 1:1 {@link MapFunction} to {@link MessageEnvelope}s in this {@link MessageStream} and returns the
+ * transformed {@link MessageStream}.
*
- * @param <A> the type of input {@code a}
- * @param <B> the type of input {@code b}
- * @param <C> the type of input {@code c}
+ * @param mapFn the function to transform a {@link MessageEnvelope} to another {@link MessageEnvelope}
+ * @param <TM> the type of {@link MessageEnvelope}s in the transformed {@link MessageStream}
+ * @return the transformed {@link MessageStream}
*/
- @FunctionalInterface
- public interface VoidFunction3<A, B, C> {
- public void apply(A a, B b, C c);
- }
+ <TM extends MessageEnvelope> MessageStream<TM> map(MapFunction<M, TM> mapFn);
/**
- * Method to apply a map function (1:1) on a {@link MessageStream}
+ * Applies the provided 1:n {@link FlatMapFunction} to transform a {@link MessageEnvelope} in this {@link MessageStream}
+ * to n {@link MessageEnvelope}s in the transformed {@link MessageStream}
*
- * @param mapper the mapper function to map one input {@link Message} to one output {@link Message}
- * @param <OM> the type of the output {@link Message} in the output {@link MessageStream}
- * @return the output {@link MessageStream} by applying the map function on the input {@link MessageStream}
+ * @param flatMapFn the function to transform a {@link MessageEnvelope} to zero or more {@link MessageEnvelope}s
+ * @param <TM> the type of {@link MessageEnvelope}s in the transformed {@link MessageStream}
+ * @return the transformed {@link MessageStream}
*/
- public <OM extends Message> MessageStream<OM> map(Function<M, OM> mapper) {
- Operator<OM> op = Operators.<M, OM>getStreamOperator(m -> new ArrayList<OM>() { {
- OM r = mapper.apply(m);
- if (r != null) {
- this.add(r);
- }
- } });
- this.subscribers.add(op);
- return op.getOutputStream();
- }
+ <TM extends MessageEnvelope> MessageStream<TM> flatMap(FlatMapFunction<M, TM> flatMapFn);
/**
- * Method to apply a flatMap function (1:n) on a {@link MessageStream}
+ * Applies the provided {@link FilterFunction} to {@link MessageEnvelope}s in this {@link MessageStream} and returns the
+ * transformed {@link MessageStream}.
+ * <p>
+ * The {@link FilterFunction} is a predicate which determines whether a {@link MessageEnvelope} in this {@link MessageStream}
+ * should be retained in the transformed {@link MessageStream}.
*
- * @param flatMapper the flat mapper function to map one input {@link Message} to zero or more output {@link Message}s
- * @param <OM> the type of the output {@link Message} in the output {@link MessageStream}
- * @return the output {@link MessageStream} by applying the map function on the input {@link MessageStream}
+ * @param filterFn the predicate to filter {@link MessageEnvelope}s from this {@link MessageStream}
+ * @return the transformed {@link MessageStream}
*/
- public <OM extends Message> MessageStream<OM> flatMap(Function<M, Collection<OM>> flatMapper) {
- Operator<OM> op = Operators.getStreamOperator(flatMapper);
- this.subscribers.add(op);
- return op.getOutputStream();
- }
+ MessageStream<M> filter(FilterFunction<M> filterFn);
/**
- * Method to apply a filter function on a {@link MessageStream}
+ * Allows sending {@link MessageEnvelope}s in this {@link MessageStream} to an output
+ * {@link org.apache.samza.system.SystemStream} using the provided {@link SinkFunction}.
*
- * @param filter the filter function to filter input {@link Message}s from the input {@link MessageStream}
- * @return the output {@link MessageStream} after applying the filter function on the input {@link MessageStream}
+ * @param sinkFn the function to send {@link MessageEnvelope}s in this stream to output systems
*/
- public MessageStream<M> filter(Function<M, Boolean> filter) {
- Operator<M> op = Operators.<M, M>getStreamOperator(t -> new ArrayList<M>() { {
- if (filter.apply(t)) {
- this.add(t);
- }
- } });
- this.subscribers.add(op);
- return op.getOutputStream();
- }
+ void sink(SinkFunction<M> sinkFn);
/**
- * Method to send an input {@link MessageStream} to an output {@link org.apache.samza.system.SystemStream}, and allows the output {@link MessageStream}
- * to be consumed by downstream stream operators again.
+ * Groups the {@link MessageEnvelope}s in this {@link MessageStream} according to the provided {@link Window} semantics
+ * (e.g. tumbling, sliding or session windows) and returns the transformed {@link MessageStream} of
+ * {@link WindowOutput}s.
+ * <p>
+ * Use the {@link org.apache.samza.operators.windows.Windows} helper methods to create the appropriate windows.
*
- * @param sink the user-defined sink function to send the input {@link Message}s to the external output systems
+ * @param window the {@link Window} to group and process {@link MessageEnvelope}s from this {@link MessageStream}
+ * @param <WK> the type of key in the {@link WindowOutput} from the {@link Window}
+ * @param <WV> the type of value in the {@link WindowOutput} from the {@link Window}
+ * @param <WS> the type of window state kept in the {@link Window}
+ * @param <WM> the type of {@link WindowOutput} in the transformed {@link MessageStream}
+ * @return the transformed {@link MessageStream}
*/
- public void sink(VoidFunction3<M, MessageCollector, TaskCoordinator> sink) {
- this.subscribers.add(Operators.getSinkOperator(sink));
- }
+ <WK, WV, WS extends WindowState<WV>, WM extends WindowOutput<WK, WV>> MessageStream<WM> window(
+ Window<M, WK, WV, WM> window);
/**
- * Method to perform a window function (i.e. a group-by, aggregate function) on a {@link MessageStream}
+ * Joins this {@link MessageStream} with another {@link MessageStream} using the provided pairwise {@link JoinFunction}.
+ * <p>
+ * We currently only support 2-way joins.
*
- * @param window the window function to group and aggregate the input {@link Message}s from the input {@link MessageStream}
- * @param <WK> the type of key in the output {@link Message} from the {@link Windows.Window} function
- * @param <WV> the type of output value from
- * @param <WS> the type of window state kept in the {@link Windows.Window} function
- * @param <WM> the type of {@link org.apache.samza.operators.internal.WindowOutput} message from the {@link Windows.Window} function
- * @return the output {@link MessageStream} after applying the window function on the input {@link MessageStream}
- */
- public <WK, WV, WS extends WindowState<WV>, WM extends WindowOutput<WK, WV>> MessageStream<WM> window(Windows.Window<M, WK, WV, WM> window) {
- Operator<WM> wndOp = Operators.getWindowOperator(Windows.getInternalWindowFn(window));
- this.subscribers.add(wndOp);
- return wndOp.getOutputStream();
- }
-
- /**
- * Method to add an input {@link MessageStream} to a join function. Note that we currently only support 2-way joins.
- *
- * @param other the other stream to be joined w/
- * @param merger the common function to merge messages from this {@link MessageStream} and {@code other}
+ * @param otherStream the other {@link MessageStream} to be joined with
+ * @param joinFn the function to join {@link MessageEnvelope}s from this and the other {@link MessageStream}
* @param <K> the type of join key
- * @param <JM> the type of message in the {@link Message} from the other join stream
- * @param <RM> the type of message in the {@link Message} from the join function
- * @return the output {@link MessageStream} from the join function {@code joiner}
+ * @param <OM> the type of {@link MessageEnvelope}s in the other stream
+ * @param <RM> the type of {@link MessageEnvelope}s resulting from the {@code joinFn}
+ * @return the joined {@link MessageStream}
*/
- public <K, JM extends Message<K, ?>, RM extends Message> MessageStream<RM> join(MessageStream<JM> other,
- BiFunction<M, JM, RM> merger) {
- MessageStream<RM> outputStream = new MessageStream<>();
-
- BiFunction<M, JM, RM> parJoin1 = merger::apply;
- BiFunction<JM, M, RM> parJoin2 = (m, t1) -> merger.apply(t1, m);
-
- // TODO: need to add default store functions for the two partial join functions
-
- other.subscribers.add(Operators.<JM, K, M, RM>getPartialJoinOperator(parJoin2, outputStream));
- this.subscribers.add(Operators.<M, K, JM, RM>getPartialJoinOperator(parJoin1, outputStream));
- return outputStream;
- }
+ <K, OM extends MessageEnvelope<K, ?>, RM extends MessageEnvelope> MessageStream<RM> join(MessageStream<OM> otherStream,
+ JoinFunction<M, OM, RM> joinFn);
/**
- * Method to merge all {@code others} streams w/ this {@link MessageStream}. The merging streams must have the same type {@code M}
+ * Merge all {@code otherStreams} with this {@link MessageStream}.
+ * <p>
+ * The merging streams must have the same {@link MessageEnvelope} type {@code M}.
*
- * @param others other streams to be merged w/ this one
- * @return the merged output stream
+ * @param otherStreams other {@link MessageStream}s to be merged with this {@link MessageStream}
+ * @return the merged {@link MessageStream}
*/
- public MessageStream<M> merge(Collection<MessageStream<M>> others) {
- MessageStream<M> outputStream = new MessageStream<>();
-
- others.add(this);
- others.forEach(other -> other.subscribers.add(Operators.getMergeOperator(outputStream)));
- return outputStream;
- }
-
+ MessageStream<M> merge(Collection<MessageStream<M>> otherStreams);
+
}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/MessageStreams.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/MessageStreams.java b/samza-api/src/main/java/org/apache/samza/operators/MessageStreams.java
deleted file mode 100644
index 51bf482..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/MessageStreams.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.operators;
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.data.IncomingSystemMessage;
-import org.apache.samza.system.SystemStreamPartition;
-
-
-/**
- * This class defines all methods to create a {@link MessageStream} object. Users can use this to create an {@link MessageStream}
- * from a specific input source.
- *
- */
-@InterfaceStability.Unstable
-public final class MessageStreams {
-
- /**
- * private constructor to prevent instantiation
- */
- private MessageStreams() {}
-
- /**
- * private class for system input/output {@link MessageStream}
- */
- public static final class SystemMessageStream extends MessageStream<IncomingSystemMessage> {
- /**
- * The corresponding {@link org.apache.samza.system.SystemStream}
- */
- private final SystemStreamPartition ssp;
-
- /**
- * Constructor for input system stream
- *
- * @param ssp the input {@link SystemStreamPartition} for the input {@link SystemMessageStream}
- */
- private SystemMessageStream(SystemStreamPartition ssp) {
- this.ssp = ssp;
- }
-
- /**
- * Getter for the {@link SystemStreamPartition} of the input
- *
- * @return the input {@link SystemStreamPartition}
- */
- public SystemStreamPartition getSystemStreamPartition() {
- return this.ssp;
- }
- }
-
- /**
- * Public static API methods start here
- */
-
- /**
- * Static API method to create a {@link MessageStream} from a system input stream
- *
- * @param ssp the input {@link SystemStreamPartition}
- * @return the {@link MessageStream} object takes {@code ssp} as the input
- */
- public static SystemMessageStream input(SystemStreamPartition ssp) {
- return new SystemMessageStream(ssp);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/StreamOperatorTask.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/StreamOperatorTask.java b/samza-api/src/main/java/org/apache/samza/operators/StreamOperatorTask.java
new file mode 100644
index 0000000..16cf27a
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/StreamOperatorTask.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.data.IncomingSystemMessageEnvelope;
+import org.apache.samza.system.SystemStreamPartition;
+
+import java.util.Map;
+
+
+/**
+ * A {@link StreamOperatorTask} is the basic interface to implement for processing {@link MessageStream}s.
+ * Implementations can describe the transformation steps for each {@link MessageStream} in the
+ * {@link #transform} method using {@link MessageStream} APIs.
+ * <p>
+ * Implementations may be augmented by implementing {@link org.apache.samza.task.InitableTask},
+ * {@link org.apache.samza.task.WindowableTask} and {@link org.apache.samza.task.ClosableTask} interfaces,
+ * but should not implement {@link org.apache.samza.task.StreamTask} or {@link org.apache.samza.task.AsyncStreamTask}
+ * interfaces.
+ */
+@InterfaceStability.Unstable
+public interface StreamOperatorTask {
+
+ /**
+ * Describe the transformation steps for each {@link MessageStream}s for this task using the
+ * {@link MessageStream} APIs. Each {@link MessageStream} corresponds to one {@link SystemStreamPartition}
+ * in the input system.
+ *
+ * @param messageStreams the {@link MessageStream}s that receive {@link IncomingSystemMessageEnvelope}s
+ * from their corresponding {@link org.apache.samza.system.SystemStreamPartition}
+ */
+ void transform(Map<SystemStreamPartition, MessageStream<IncomingSystemMessageEnvelope>> messageStreams);
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/TriggerBuilder.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/TriggerBuilder.java b/samza-api/src/main/java/org/apache/samza/operators/TriggerBuilder.java
deleted file mode 100644
index 5b3f4d0..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/TriggerBuilder.java
+++ /dev/null
@@ -1,323 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.operators;
-
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.data.Message;
-import org.apache.samza.operators.internal.Trigger;
-
-import java.util.concurrent.TimeUnit;
-import java.util.function.BiFunction;
-import java.util.function.Function;
-
-
-/**
- * This class defines a builder of {@link org.apache.samza.operators.internal.Trigger} object for a {@link Windows.Window}. The triggers are categorized into
- * three types:
- *
- * <p>
- * early trigger: defines the condition when the first output from the window function is sent.
- * late trigger: defines the condition when the updated output after the first output is sent.
- * timer trigger: defines a system timeout condition to trigger output if no more inputs are received to enable early/late triggers
- * </p>
- *
- * If multiple conditions are defined for a specific type of trigger, the aggregated trigger is the disjunction of the each individual trigger (i.e. OR).
- *
- * NOTE: Programmers should not use classes defined in {@link org.apache.samza.operators.internal} to create triggers
- *
- *
- * @param <M> the type of input {@link Message} to the {@link Windows.Window}
- * @param <V> the type of output value from the {@link Windows.Window}
- */
-@InterfaceStability.Unstable
-public final class TriggerBuilder<M extends Message, V> {
-
- /**
- * Predicate helper to OR multiple trigger conditions
- */
- static class PredicateHelper {
- static <M, S> BiFunction<M, S, Boolean> or(BiFunction<M, S, Boolean> lhs, BiFunction<M, S, Boolean> rhs) {
- return (m, s) -> lhs.apply(m, s) || rhs.apply(m, s);
- }
-
- static <S> Function<S, Boolean> or(Function<S, Boolean> lhs, Function<S, Boolean> rhs) {
- return s -> lhs.apply(s) || rhs.apply(s);
- }
- }
-
- /**
- * The early trigger condition that determines the first output from the {@link Windows.Window}
- */
- private BiFunction<M, WindowState<V>, Boolean> earlyTrigger = null;
-
- /**
- * The late trigger condition that determines the late output(s) from the {@link Windows.Window}
- */
- private BiFunction<M, WindowState<V>, Boolean> lateTrigger = null;
-
- /**
- * The system timer based trigger conditions that guarantees the {@link Windows.Window} proceeds forward
- */
- private Function<WindowState<V>, Boolean> timerTrigger = null;
-
- /**
- * The state updater function to be applied after the first output is triggered
- */
- private Function<WindowState<V>, WindowState<V>> earlyTriggerUpdater = Function.identity();
-
- /**
- * The state updater function to be applied after the late output is triggered
- */
- private Function<WindowState<V>, WindowState<V>> lateTriggerUpdater = Function.identity();
-
- /**
- * Helper method to add a trigger condition
- *
- * @param currentTrigger current trigger condition
- * @param newTrigger new trigger condition
- * @return combined trigger condition that is {@code currentTrigger} OR {@code newTrigger}
- */
- private BiFunction<M, WindowState<V>, Boolean> addTrigger(BiFunction<M, WindowState<V>, Boolean> currentTrigger,
- BiFunction<M, WindowState<V>, Boolean> newTrigger) {
- if (currentTrigger == null) {
- return newTrigger;
- }
-
- return PredicateHelper.or(currentTrigger, newTrigger);
- }
-
- /**
- * Helper method to add a system timer trigger
- *
- * @param currentTimer current timer condition
- * @param newTimer new timer condition
- * @return combined timer condition that is {@code currentTimer} OR {@code newTimer}
- */
- private Function<WindowState<V>, Boolean> addTimerTrigger(Function<WindowState<V>, Boolean> currentTimer,
- Function<WindowState<V>, Boolean> newTimer) {
- if (currentTimer == null) {
- return newTimer;
- }
-
- return PredicateHelper.or(currentTimer, newTimer);
- }
-
- /**
- * default constructor to prevent instantiation
- */
- private TriggerBuilder() {}
-
- /**
- * Constructor that set the size limit as the early trigger for a window
- *
- * @param sizeLimit the number of messages in a window that would trigger the first output
- */
- private TriggerBuilder(long sizeLimit) {
- this.earlyTrigger = (m, s) -> s.getNumberMessages() > sizeLimit;
- }
-
- /**
- * Constructor that set the event time length as the early trigger
- *
- * @param eventTimeFunction the function that calculate the event time in nano-second from the input {@link Message}
- * @param wndLenMs the window length in event time in milli-second
- */
- private TriggerBuilder(Function<M, Long> eventTimeFunction, long wndLenMs) {
- this.earlyTrigger = (m, s) ->
- TimeUnit.NANOSECONDS.toMillis(Math.max(s.getLatestEventTimeNs() - s.getEarliestEventTimeNs(),
- eventTimeFunction.apply(m) - s.getEarliestEventTimeNs())) > wndLenMs;
- }
-
- /**
- * Constructor that set the special token message as the early trigger
- *
- * @param tokenFunc the function that checks whether an input {@link Message} is a token message that triggers window output
- */
- private TriggerBuilder(Function<M, Boolean> tokenFunc) {
- this.earlyTrigger = (m, s) -> tokenFunc.apply(m);
- }
-
- /**
- * Build method that creates an {@link org.apache.samza.operators.internal.Trigger} object based on the trigger conditions set in {@link TriggerBuilder}
- * This is kept package private and only used by {@link Windows} to convert the mutable {@link TriggerBuilder} object to an immutable {@link Trigger} object
- *
- * @return the final {@link org.apache.samza.operators.internal.Trigger} object
- */
- Trigger<M, WindowState<V>> build() {
- return Trigger.createTrigger(this.timerTrigger, this.earlyTrigger, this.lateTrigger, this.earlyTriggerUpdater, this.lateTriggerUpdater);
- }
-
- /**
- * Public API methods start here
- */
-
-
- /**
- * API method to allow users to set an update method to update the output value after the first window output is triggered
- * by the early trigger condition
- *
- * @param onTriggerFunc the method to update the output value after the early trigger
- * @return the {@link TriggerBuilder} object
- */
- public TriggerBuilder<M, V> onEarlyTrigger(Function<V, V> onTriggerFunc) {
- this.earlyTriggerUpdater = s -> {
- s.setOutputValue(onTriggerFunc.apply(s.getOutputValue()));
- return s;
- };
- return this;
- }
-
- /**
- * API method to allow users to set an update method to update the output value after a late window output is triggered
- * by the late trigger condition
- *
- * @param onTriggerFunc the method to update the output value after the late trigger
- * @return the {@link TriggerBuilder} object
- */
- public TriggerBuilder<M, V> onLateTrigger(Function<V, V> onTriggerFunc) {
- this.lateTriggerUpdater = s -> {
- s.setOutputValue(onTriggerFunc.apply(s.getOutputValue()));
- return s;
- };
- return this;
- }
-
- /**
- * API method to allow users to add a system timer trigger based on timeout after the last message received in the window
- *
- * @param timeoutMs the timeout in ms after the last message received in the window
- * @return the {@link TriggerBuilder} object
- */
- public TriggerBuilder<M, V> addTimeoutSinceLastMessage(long timeoutMs) {
- this.timerTrigger = this.addTimerTrigger(this.timerTrigger,
- s -> TimeUnit.NANOSECONDS.toMillis(s.getLastMessageTimeNs()) + timeoutMs < System.currentTimeMillis());
- return this;
- }
-
- /**
- * API method to allow users to add a system timer trigger based on the timeout after the first message received in the window
- *
- * @param timeoutMs the timeout in ms after the first message received in the window
- * @return the {@link TriggerBuilder} object
- */
- public TriggerBuilder<M, V> addTimeoutSinceFirstMessage(long timeoutMs) {
- this.timerTrigger = this.addTimerTrigger(this.timerTrigger, s ->
- TimeUnit.NANOSECONDS.toMillis(s.getFirstMessageTimeNs()) + timeoutMs < System.currentTimeMillis());
- return this;
- }
-
- /**
- * API method allow users to add a late trigger based on the window size limit
- *
- * @param sizeLimit limit on the number of messages in window
- * @return the {@link TriggerBuilder} object
- */
- public TriggerBuilder<M, V> addLateTriggerOnSizeLimit(long sizeLimit) {
- this.lateTrigger = this.addTrigger(this.lateTrigger, (m, s) -> s.getNumberMessages() > sizeLimit);
- return this;
- }
-
- /**
- * API method to allow users to define a customized late trigger function based on input message and the window state
- *
- * @param lateTrigger the late trigger condition based on input {@link Message} and the current {@link WindowState}
- * @return the {@link TriggerBuilder} object
- */
- public TriggerBuilder<M, V> addLateTrigger(BiFunction<M, WindowState<V>, Boolean> lateTrigger) {
- this.lateTrigger = this.addTrigger(this.lateTrigger, lateTrigger);
- return this;
- }
-
- /**
- * Static API method to create a {@link TriggerBuilder} w/ early trigger condition based on window size limit
- *
- * @param sizeLimit window size limit
- * @param <M> the type of input {@link Message}
- * @param <V> the type of {@link Windows.Window} output value
- * @return the {@link TriggerBuilder} object
- */
- public static <M extends Message, V> TriggerBuilder<M, V> earlyTriggerWhenExceedWndLen(long sizeLimit) {
- return new TriggerBuilder<M, V>(sizeLimit);
- }
-
- /**
- * Static API method to create a {@link TriggerBuilder} w/ early trigger condition based on event time window
- *
- *
- * @param eventTimeFunc the function to get the event time from the input message
- * @param eventTimeWndSizeMs the event time window size in Ms
- * @param <M> the type of input {@link Message}
- * @param <V> the type of {@link Windows.Window} output value
- * @return the {@link TriggerBuilder} object
- */
- public static <M extends Message, V> TriggerBuilder<M, V> earlyTriggerOnEventTime(Function<M, Long> eventTimeFunc, long eventTimeWndSizeMs) {
- return new TriggerBuilder<M, V>(eventTimeFunc, eventTimeWndSizeMs);
- }
-
- /**
- * Static API method to create a {@link TriggerBuilder} w/ early trigger condition based on token messages
- *
- * @param tokenFunc the function to determine whether an input message is a window token or not
- * @param <M> the type of input {@link Message}
- * @param <V> the type of {@link Windows.Window} output value
- * @return the {@link TriggerBuilder} object
- */
- public static <M extends Message, V> TriggerBuilder<M, V> earlyTriggerOnTokenMsg(Function<M, Boolean> tokenFunc) {
- return new TriggerBuilder<M, V>(tokenFunc);
- }
-
- /**
- * Static API method to allow customized early trigger condition based on input {@link Message} and the corresponding {@link WindowState}
- *
- * @param earlyTrigger the user defined early trigger condition
- * @param <M> the input message type
- * @param <V> the output value from the window
- * @return the {@link TriggerBuilder} object
- */
- public static <M extends Message, V> TriggerBuilder<M, V> earlyTrigger(BiFunction<M, WindowState<V>, Boolean> earlyTrigger) {
- TriggerBuilder<M, V> newTriggers = new TriggerBuilder<M, V>();
- newTriggers.earlyTrigger = newTriggers.addTrigger(newTriggers.earlyTrigger, earlyTrigger);
- return newTriggers;
- }
-
- /**
- * Static API method to create a {@link TriggerBuilder} w/ system timeout after the last message received in the window
- *
- * @param timeoutMs timeout in ms after the last message received
- * @param <M> the type of input {@link Message}
- * @param <V> the type of {@link Windows.Window} output value
- * @return the {@link TriggerBuilder} object
- */
- public static <M extends Message, V> TriggerBuilder<M, V> timeoutSinceLastMessage(long timeoutMs) {
- return new TriggerBuilder<M, V>().addTimeoutSinceLastMessage(timeoutMs);
- }
-
- /**
- * Static API method to create a {@link TriggerBuilder} w/ system timeout after the first message received in the window
- *
- * @param timeoutMs timeout in ms after the first message received
- * @param <M> the type of input {@link Message}
- * @param <V> the type of {@link Windows.Window} output value
- * @return the {@link TriggerBuilder} object
- */
- public static <M extends Message, V> TriggerBuilder<M, V> timeoutSinceFirstMessage(long timeoutMs) {
- return new TriggerBuilder<M, V>().addTimeoutSinceFirstMessage(timeoutMs);
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/WindowState.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/WindowState.java b/samza-api/src/main/java/org/apache/samza/operators/WindowState.java
deleted file mode 100644
index 759f2d8..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/WindowState.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.operators;
-
-import org.apache.samza.annotation.InterfaceStability;
-
-
-/**
- * This interface defines the methods a window state class has to implement. The programmers are allowed to implement
- * customized window state to be stored in window state stores by implementing this interface class.
- *
- * @param <WV> the type for window output value
- */
-@InterfaceStability.Unstable
-public interface WindowState<WV> {
- /**
- * Method to get the system time when the first message in the window is received
- *
- * @return nano-second of system time for the first message received in the window
- */
- long getFirstMessageTimeNs();
-
- /**
- * Method to get the system time when the last message in the window is received
- *
- * @return nano-second of system time for the last message received in the window
- */
- long getLastMessageTimeNs();
-
- /**
- * Method to get the earliest event time in the window
- *
- * @return the earliest event time in nano-second in the window
- */
- long getEarliestEventTimeNs();
-
- /**
- * Method to get the latest event time in the window
- *
- * @return the latest event time in nano-second in the window
- */
- long getLatestEventTimeNs();
-
- /**
- * Method to get the total number of messages received in the window
- *
- * @return number of messages in the window
- */
- long getNumberMessages();
-
- /**
- * Method to get the corresponding window's output value
- *
- * @return the corresponding window's output value
- */
- WV getOutputValue();
-
- /**
- * Method to set the corresponding window's output value
- *
- * @param value the corresponding window's output value
- */
- void setOutputValue(WV value);
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/Windows.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/Windows.java b/samza-api/src/main/java/org/apache/samza/operators/Windows.java
deleted file mode 100644
index 6619f41..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/Windows.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.operators;
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.data.Message;
-import org.apache.samza.operators.internal.Operators;
-import org.apache.samza.operators.internal.Trigger;
-import org.apache.samza.operators.internal.WindowFn;
-import org.apache.samza.operators.internal.WindowOutput;
-import org.apache.samza.storage.kv.Entry;
-
-import java.util.Collection;
-import java.util.function.BiFunction;
-import java.util.function.Function;
-
-
-/**
- * This class defines a collection of {@link Window} functions. The public classes and methods here are intended to be
- * used by the user (i.e. programmers) to create {@link Window} function directly.
- *
- */
-@InterfaceStability.Unstable
-public final class Windows {
-
- /**
- * private constructor to prevent instantiation
- */
- private Windows() {}
-
- /**
- * This class defines a session window function class
- *
- * @param <M> the type of input {@link Message}
- * @param <WK> the type of session key in the session window
- * @param <WV> the type of output value in each session window
- */
- static class SessionWindow<M extends Message, WK, WV> implements Window<M, WK, WV, WindowOutput<WK, WV>> {
-
- /**
- * Constructor. Made private s.t. it can only be instantiated via the static API methods in {@link Windows}
- *
- * @param sessionKeyFunction function to get the session key from the input {@link Message}
- * @param aggregator function to calculate the output value based on the input {@link Message} and current output value
- */
- private SessionWindow(Function<M, WK> sessionKeyFunction, BiFunction<M, WV, WV> aggregator) {
- this.wndKeyFunction = sessionKeyFunction;
- this.aggregator = aggregator;
- }
-
- /**
- * function to calculate the window key from input message
- */
- private final Function<M, WK> wndKeyFunction;
-
- /**
- * function to calculate the output value from the input message and the current output value
- */
- private final BiFunction<M, WV, WV> aggregator;
-
- /**
- * trigger condition that determines when to send out the output value in a {@link WindowOutput} message
- */
- private Trigger<M, WindowState<WV>> trigger = null;
-
- //TODO: need to create a set of {@link StoreFunctions} that is default to input {@link Message} type for {@link Window}
- private Operators.StoreFunctions<M, WK, WindowState<WV>> storeFunctions = null;
-
- /**
- * Public API methods start here
- */
-
- /**
- * Public API method to define the watermark trigger for the window operator
- *
- * @param wndTrigger {@link Trigger} function defines the watermark trigger for this {@link SessionWindow}
- * @return The window operator w/ the defined watermark trigger
- */
- @Override
- public Window<M, WK, WV, WindowOutput<WK, WV>> setTriggers(TriggerBuilder<M, WV> wndTrigger) {
- this.trigger = wndTrigger.build();
- return this;
- }
-
- private BiFunction<M, Entry<WK, WindowState<WV>>, WindowOutput<WK, WV>> getTransformFunc() {
- // TODO: actual implementation of the main session window logic, based on the wndKeyFunction, aggregator, and triggers;
- return null;
- }
-
- private WindowFn<M, WK, WindowState<WV>, WindowOutput<WK, WV>> getInternalWindowFn() {
- return new WindowFn<M, WK, WindowState<WV>, WindowOutput<WK, WV>>() {
-
- @Override public BiFunction<M, Entry<WK, WindowState<WV>>, WindowOutput<WK, WV>> getTransformFunc() {
- return SessionWindow.this.getTransformFunc();
- }
-
- @Override public Operators.StoreFunctions<M, WK, WindowState<WV>> getStoreFuncs() {
- return SessionWindow.this.storeFunctions;
- }
-
- @Override public Trigger<M, WindowState<WV>> getTrigger() {
- return SessionWindow.this.trigger;
- }
- };
- }
- }
-
- static <M extends Message, WK, WV, WS extends WindowState<WV>, WM extends WindowOutput<WK, WV>> WindowFn<M, WK, WS, WM> getInternalWindowFn(
- Window<M, WK, WV, WM> window) {
- if (window instanceof SessionWindow) {
- SessionWindow<M, WK, WV> sessionWindow = (SessionWindow<M, WK, WV>) window;
- return (WindowFn<M, WK, WS, WM>) sessionWindow.getInternalWindowFn();
- }
- throw new IllegalArgumentException("Input window type not supported.");
- }
-
- /**
- * Public static API methods start here
- *
- */
-
- /**
- * The public programming interface class for window function
- *
- * @param <M> the type of input {@link Message}
- * @param <WK> the type of key to the {@link Window}
- * @param <WV> the type of output value in the {@link WindowOutput}
- * @param <WM> the type of message in the window output stream
- */
- @InterfaceStability.Unstable
- public interface Window<M extends Message, WK, WV, WM extends WindowOutput<WK, WV>> {
-
- /**
- * Set the triggers for this {@link Window}
- *
- * @param wndTrigger trigger conditions set by the programmers
- * @return the {@link Window} function w/ the trigger {@code wndTrigger}
- */
- Window<M, WK, WV, WM> setTriggers(TriggerBuilder<M, WV> wndTrigger);
- }
-
- /**
- * Static API method to create a {@link SessionWindow} in which the output value is simply the collection of input messages
- *
- * @param sessionKeyFunction function to calculate session window key
- * @param <M> type of input {@link Message}
- * @param <WK> type of the session window key
- * @return the {@link Window} function for the session
- */
- public static <M extends Message, WK> Window<M, WK, Collection<M>, WindowOutput<WK, Collection<M>>> intoSessions(Function<M, WK> sessionKeyFunction) {
- return new SessionWindow<>(sessionKeyFunction, (m, c) -> {
- c.add(m);
- return c;
- });
- }
-
- /**
- * Static API method to create a {@link SessionWindow} in which the output value is a collection of {@code SI} from the input messages
- *
- * @param sessionKeyFunction function to calculate session window key
- * @param sessionInfoExtractor function to retrieve session info of type {@code SI} from the input message of type {@code M}
- * @param <M> type of the input {@link Message}
- * @param <WK> type of the session window key
- * @param <SI> type of the session information retrieved from each input message of type {@code M}
- * @return the {@link Window} function for the session
- */
- public static <M extends Message, WK, SI> Window<M, WK, Collection<SI>, WindowOutput<WK, Collection<SI>>> intoSessions(Function<M, WK> sessionKeyFunction,
- Function<M, SI> sessionInfoExtractor) {
- return new SessionWindow<>(sessionKeyFunction, (m, c) -> {
- c.add(sessionInfoExtractor.apply(m));
- return c;
- });
- }
-
- /**
- * Static API method to create a {@link SessionWindow} as a counter of input messages
- *
- * @param sessionKeyFunction function to calculate session window key
- * @param <M> type of the input {@link Message}
- * @param <WK> type of the session window key
- * @return the {@link Window} function for the session
- */
- public static <M extends Message, WK> Window<M, WK, Integer, WindowOutput<WK, Integer>> intoSessionCounter(Function<M, WK> sessionKeyFunction) {
- return new SessionWindow<>(sessionKeyFunction, (m, c) -> c + 1);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessage.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessage.java b/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessage.java
deleted file mode 100644
index 3c9874d..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessage.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.operators.data;
-
-import org.apache.samza.system.IncomingMessageEnvelope;
-import org.apache.samza.system.SystemStreamPartition;
-
-
-/**
- * This class implements a {@link Message} that encapsulates an {@link IncomingMessageEnvelope} from the system
- *
- */
-public class IncomingSystemMessage implements Message<Object, Object>, InputSystemMessage<Offset> {
- /**
- * Incoming message envelope
- */
- private final IncomingMessageEnvelope imsg;
-
- /**
- * The receive time of this incoming message
- */
- private final long recvTimeNano;
-
- /**
- * Ctor to create a {@code IncomingSystemMessage} from {@link IncomingMessageEnvelope}
- *
- * @param imsg The incoming system message
- */
- public IncomingSystemMessage(IncomingMessageEnvelope imsg) {
- this.imsg = imsg;
- this.recvTimeNano = System.nanoTime();
- }
-
- @Override
- public Object getMessage() {
- return this.imsg.getMessage();
- }
-
- @Override
- public Object getKey() {
- return this.imsg.getKey();
- }
-
- @Override
- public long getTimestamp() {
- return this.recvTimeNano;
- }
-
- @Override
- public Offset getOffset() {
- // TODO: need to add offset factory to generate different types of offset. This is just a placeholder,
- // assuming incoming message carries long value as offset (i.e. Kafka case)
- return new LongOffset(this.imsg.getOffset());
- }
-
- @Override
- public SystemStreamPartition getSystemStreamPartition() {
- return imsg.getSystemStreamPartition();
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessageEnvelope.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessageEnvelope.java b/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessageEnvelope.java
new file mode 100644
index 0000000..a65809c
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/data/IncomingSystemMessageEnvelope.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators.data;
+
+import org.apache.samza.system.IncomingMessageEnvelope;
+import org.apache.samza.system.SystemStreamPartition;
+
+
+/**
+ * A {@link MessageEnvelope} that provides additional information about its input {@link SystemStreamPartition}
+ * and its {@link Offset} within the {@link SystemStreamPartition}.
+ * <p>
+ * Note: the {@link Offset} is only unique and comparable within its {@link SystemStreamPartition}.
+ */
+public class IncomingSystemMessageEnvelope implements MessageEnvelope<Object, Object> {
+
+ private final IncomingMessageEnvelope ime;
+
+ /**
+ * Creates an {@code IncomingSystemMessageEnvelope} from the {@link IncomingMessageEnvelope}.
+ *
+ * @param ime the {@link IncomingMessageEnvelope} from the input system.
+ */
+ public IncomingSystemMessageEnvelope(IncomingMessageEnvelope ime) {
+ this.ime = ime;
+ }
+
+ @Override
+ public Object getKey() {
+ return this.ime.getKey();
+ }
+
+ @Override
+ public Object getMessage() {
+ return this.ime.getMessage();
+ }
+
+ public Offset getOffset() {
+ // TODO: need to add offset factory to generate different types of offset. This is just a placeholder,
+ // assuming incoming message envelope carries long value as offset (i.e. Kafka case)
+ return new LongOffset(this.ime.getOffset());
+ }
+
+ public SystemStreamPartition getSystemStreamPartition() {
+ return this.ime.getSystemStreamPartition();
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/data/InputSystemMessage.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/data/InputSystemMessage.java b/samza-api/src/main/java/org/apache/samza/operators/data/InputSystemMessage.java
deleted file mode 100644
index 5c23e74..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/data/InputSystemMessage.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.operators.data;
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.system.SystemStreamPartition;
-
-
-/**
- * This interface defines additional methods a message from an system input should implement, including the methods to
- * get {@link SystemStreamPartition} and the {@link Offset} of the input system message.
- */
-@InterfaceStability.Unstable
-public interface InputSystemMessage<O extends Offset> {
-
- /**
- * Get the input message's {@link SystemStreamPartition}
- *
- * @return the {@link SystemStreamPartition} this message is coming from
- */
- SystemStreamPartition getSystemStreamPartition();
-
- /**
- * Get the offset of the message in the input stream. This should be used to uniquely identify a message in an input stream.
- *
- * @return The offset of the message in the input stream.
- */
- O getOffset();
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/data/Message.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/data/Message.java b/samza-api/src/main/java/org/apache/samza/operators/data/Message.java
deleted file mode 100644
index 8441682..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/data/Message.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.operators.data;
-
-import org.apache.samza.annotation.InterfaceStability;
-
-
-/**
- * This class defines the generic interface of {@link Message}, which is a entry in the input/output stream.
- *
- * <p>The {@link Message} models the basic operatible unit in streaming SQL processes in Samza.
- *
- */
-@InterfaceStability.Unstable
-public interface Message<K, M> {
-
- /**
- * Access method to get the corresponding message body in {@link Message}
- *
- * @return Message object in this {@link Message}
- */
- M getMessage();
-
- /**
- * Method to indicate whether this {@link Message} indicates deletion of a message w/ the message key
- *
- * @return A boolean value indicates whether the current message is a delete or insert message
- */
- default boolean isDelete() {
- return false;
- };
-
- /**
- * Access method to the key of the message
- *
- * @return The key of the message
- */
- K getKey();
-
- /**
- * Get the message creation timestamp of the message.
- *
- * @return The message's timestamp in nano seconds.
- */
- long getTimestamp();
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/data/MessageEnvelope.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/data/MessageEnvelope.java b/samza-api/src/main/java/org/apache/samza/operators/data/MessageEnvelope.java
new file mode 100644
index 0000000..ad64231
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/data/MessageEnvelope.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.operators.data;
+
+import org.apache.samza.annotation.InterfaceStability;
+
+
+/**
+ * An entry in the input/output {@link org.apache.samza.operators.MessageStream}s.
+ */
+@InterfaceStability.Unstable
+public interface MessageEnvelope<K, M> {
+
+ /**
+ * Get the key for this {@link MessageEnvelope}.
+ *
+ * @return the key for this {@link MessageEnvelope}
+ */
+ K getKey();
+
+ /**
+ * Get the message in this {@link MessageEnvelope}.
+ *
+ * @return the message in this {@link MessageEnvelope}
+ */
+ M getMessage();
+
+ /**
+ * Whether this {@link MessageEnvelope} indicates deletion of a previous message with this key.
+ *
+ * @return true if the current {@link MessageEnvelope} indicates deletion of a previous message with this key
+ */
+ default boolean isDelete() {
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/functions/FilterFunction.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/functions/FilterFunction.java b/samza-api/src/main/java/org/apache/samza/operators/functions/FilterFunction.java
new file mode 100644
index 0000000..e611cd0
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/functions/FilterFunction.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators.functions;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.data.MessageEnvelope;
+
+
+/**
+ * A function that specifies whether a {@link MessageEnvelope} should be retained for further processing or filtered out.
+ * @param <M> type of the input {@link MessageEnvelope}
+ */
+@InterfaceStability.Unstable
+@FunctionalInterface
+public interface FilterFunction<M extends MessageEnvelope> {
+
+ /**
+ * Returns a boolean indicating whether this {@link MessageEnvelope} should be retained or filtered out.
+ * @param message the {@link MessageEnvelope} to be checked
+ * @return true if {@link MessageEnvelope} should be retained
+ */
+ boolean apply(M message);
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/functions/FlatMapFunction.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/functions/FlatMapFunction.java b/samza-api/src/main/java/org/apache/samza/operators/functions/FlatMapFunction.java
new file mode 100644
index 0000000..dbc0bd9
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/functions/FlatMapFunction.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators.functions;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.data.MessageEnvelope;
+
+import java.util.Collection;
+
+
+/**
+ * A function that transforms a {@link MessageEnvelope} into a collection of 0 or more {@link MessageEnvelope}s,
+ * possibly of a different type.
+ * @param <M> type of the input {@link MessageEnvelope}
+ * @param <OM> type of the transformed {@link MessageEnvelope}s
+ */
+@InterfaceStability.Unstable
+@FunctionalInterface
+public interface FlatMapFunction<M extends MessageEnvelope, OM extends MessageEnvelope> {
+
+ /**
+ * Transforms the provided {@link MessageEnvelope} into a collection of 0 or more {@link MessageEnvelope}s.
+ * @param message the {@link MessageEnvelope} to be transformed
+ * @return a collection of 0 or more transformed {@link MessageEnvelope}s
+ */
+ Collection<OM> apply(M message);
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/functions/JoinFunction.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/functions/JoinFunction.java b/samza-api/src/main/java/org/apache/samza/operators/functions/JoinFunction.java
new file mode 100644
index 0000000..8cb1fce
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/functions/JoinFunction.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators.functions;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.data.MessageEnvelope;
+
+
+/**
+ * A function that joins {@link MessageEnvelope}s from two {@link org.apache.samza.operators.MessageStream}s and produces
+ * a joined {@link MessageEnvelope}.
+ * @param <M> type of the input {@link MessageEnvelope}
+ * @param <JM> type of the {@link MessageEnvelope} to join with
+ * @param <RM> type of the joined {@link MessageEnvelope}
+ */
+@InterfaceStability.Unstable
+@FunctionalInterface
+public interface JoinFunction<M extends MessageEnvelope, JM extends MessageEnvelope, RM extends MessageEnvelope> {
+
+ /**
+ * Join the provided {@link MessageEnvelope}s and produces the joined {@link MessageEnvelope}.
+ * @param message the input {@link MessageEnvelope}
+ * @param otherMessage the {@link MessageEnvelope} to join with
+ * @return the joined {@link MessageEnvelope}
+ */
+ RM apply(M message, JM otherMessage);
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/functions/MapFunction.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/functions/MapFunction.java b/samza-api/src/main/java/org/apache/samza/operators/functions/MapFunction.java
new file mode 100644
index 0000000..04919a7
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/functions/MapFunction.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators.functions;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.data.MessageEnvelope;
+
+
+/**
+ * A function that transforms a {@link MessageEnvelope} into another {@link MessageEnvelope}, possibly of a different type.
+ * @param <M> type of the input {@link MessageEnvelope}
+ * @param <OM> type of the transformed {@link MessageEnvelope}
+ */
+@InterfaceStability.Unstable
+@FunctionalInterface
+public interface MapFunction<M extends MessageEnvelope, OM extends MessageEnvelope> {
+
+ /**
+ * Transforms the provided {@link MessageEnvelope} into another {@link MessageEnvelope}
+ * @param message the {@link MessageEnvelope} to be transformed
+ * @return the transformed {@link MessageEnvelope}
+ */
+ OM apply(M message);
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/00543804/samza-api/src/main/java/org/apache/samza/operators/functions/SinkFunction.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/functions/SinkFunction.java b/samza-api/src/main/java/org/apache/samza/operators/functions/SinkFunction.java
new file mode 100644
index 0000000..505da92
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/operators/functions/SinkFunction.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.operators.functions;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.data.MessageEnvelope;
+import org.apache.samza.task.MessageCollector;
+import org.apache.samza.task.TaskCoordinator;
+
+
+/**
+ * A function that allows sending a {@link MessageEnvelope} to an output system.
+ * @param <M> type of the input {@link MessageEnvelope}
+ */
+@InterfaceStability.Unstable
+@FunctionalInterface
+public interface SinkFunction<M extends MessageEnvelope> {
+
+ /**
+ * Allows sending the provided {@link MessageEnvelope} to an output {@link org.apache.samza.system.SystemStream} using
+ * the provided {@link MessageCollector}. Also provides access to the {@link TaskCoordinator} to request commits
+ * or shut the container down.
+ *
+ * @param message the {@link MessageEnvelope} to be sent to an output {@link org.apache.samza.system.SystemStream}
+ * @param messageCollector the {@link MessageCollector} to use to send the {@link MessageEnvelope}
+ * @param taskCoordinator the {@link TaskCoordinator} to request commits or shutdown
+ */
+ void apply(M message, MessageCollector messageCollector, TaskCoordinator taskCoordinator);
+
+}