You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by xi...@apache.org on 2017/06/07 19:42:21 UTC
[3/4] samza git commit: SAMZA-1221,
SAMZA-1101: Internal cleanup for High-Level API implementation.
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/JoinOperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/JoinOperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/JoinOperatorSpec.java
new file mode 100644
index 0000000..16f59d7
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/JoinOperatorSpec.java
@@ -0,0 +1,73 @@
+/*
+ * 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.spec;
+
+import org.apache.samza.operators.functions.JoinFunction;
+
+
+/**
+ * The spec for the join operator that buffers messages from one stream and
+ * joins them with buffered messages from another stream.
+ *
+ * @param <K> the type of join key
+ * @param <M> the type of message in this stream
+ * @param <JM> the type of message in the other stream
+ * @param <RM> the type of join result
+ */
+public class JoinOperatorSpec<K, M, JM, RM> extends OperatorSpec<Object, RM> { // Object == M | JM
+
+ private final OperatorSpec<?, M> leftInputOpSpec;
+ private final OperatorSpec<?, JM> rightInputOpSpec;
+ private final JoinFunction<K, M, JM, RM> joinFn;
+ private final long ttlMs;
+
+ /**
+ * Default constructor for a {@link JoinOperatorSpec}.
+ *
+ * @param leftInputOpSpec the operator spec for the stream on the left side of the join
+ * @param rightInputOpSpec the operator spec for the stream on the right side of the join
+ * @param joinFn the user-defined join function to get join keys and results
+ * @param ttlMs the ttl in ms for retaining messages in each stream
+ * @param opId the unique ID for this operator
+ */
+ JoinOperatorSpec(OperatorSpec<?, M> leftInputOpSpec, OperatorSpec<?, JM> rightInputOpSpec,
+ JoinFunction<K, M, JM, RM> joinFn, long ttlMs, int opId) {
+ super(OpCode.JOIN, opId);
+ this.leftInputOpSpec = leftInputOpSpec;
+ this.rightInputOpSpec = rightInputOpSpec;
+ this.joinFn = joinFn;
+ this.ttlMs = ttlMs;
+ }
+
+ public OperatorSpec getLeftInputOpSpec() {
+ return leftInputOpSpec;
+ }
+
+ public OperatorSpec getRightInputOpSpec() {
+ return rightInputOpSpec;
+ }
+
+ public JoinFunction<K, M, JM, RM> getJoinFn() {
+ return this.joinFn;
+ }
+
+ public long getTtlMs() {
+ return ttlMs;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpec.java
index 3ea52ca..f64e123 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpec.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpec.java
@@ -19,19 +19,23 @@
package org.apache.samza.operators.spec;
import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.MessageStreamImpl;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Set;
/**
* A stream operator specification that holds all the information required to transform
- * the input {@link MessageStreamImpl} and produce the output {@link MessageStreamImpl}.
+ * the input {@link org.apache.samza.operators.MessageStreamImpl} and produce the output
+ * {@link org.apache.samza.operators.MessageStreamImpl}.
*
+ * @param <M> the type of input message to the operator
* @param <OM> the type of output message from the operator
*/
@InterfaceStability.Unstable
-public interface OperatorSpec<OM> {
+public abstract class OperatorSpec<M, OM> {
- enum OpCode {
+ public enum OpCode {
INPUT,
MAP,
FLAT_MAP,
@@ -41,39 +45,77 @@ public interface OperatorSpec<OM> {
JOIN,
WINDOW,
MERGE,
- PARTITION_BY
+ PARTITION_BY,
+ OUTPUT
}
+ private final int opId;
+ private final OpCode opCode;
+ private StackTraceElement[] creationStackTrace;
+
/**
- * Get the next {@link MessageStreamImpl} that receives the transformed messages produced by this operator.
- * @return the next {@link MessageStreamImpl}
+ * The set of operators that consume the messages produced from this operator.
+ * <p>
+ * We use a LinkedHashSet since we need deterministic ordering in initializing/closing operators.
*/
- MessageStreamImpl<OM> getNextStream();
+ private final Set<OperatorSpec<OM, ?>> nextOperatorSpecs = new LinkedHashSet<>();
+
+ public OperatorSpec(OpCode opCode, int opId) {
+ this.opCode = opCode;
+ this.opId = opId;
+ this.creationStackTrace = Thread.currentThread().getStackTrace();
+ }
+
+ /**
+ * Register the next operator spec in the chain that this operator should propagate its output to.
+ * @param nextOperatorSpec the next operator in the chain.
+ */
+ public void registerNextOperatorSpec(OperatorSpec<OM, ?> nextOperatorSpec) {
+ nextOperatorSpecs.add(nextOperatorSpec);
+ }
+
+ public Collection<OperatorSpec<OM, ?>> getRegisteredOperatorSpecs() {
+ return nextOperatorSpecs;
+ }
/**
* Get the {@link OpCode} for this operator.
* @return the {@link OpCode} for this operator
*/
- OpCode getOpCode();
+ public final OpCode getOpCode() {
+ return this.opCode;
+ }
/**
* Get the unique ID of this operator in the {@link org.apache.samza.operators.StreamGraph}.
* @return the unique operator ID
*/
- int getOpId();
+ public final int getOpId() {
+ return this.opId;
+ }
/**
- * Return the user source code location that creates the operator
- * @return source location
+ * Get the user source code location that created the operator.
+ * @return source code location for the operator
*/
- String getSourceLocation();
+ public final String getSourceLocation() {
+ // The stack trace for most operators looks like:
+ // [0] Thread.getStackTrace()
+ // [1] OperatorSpec.init<>()
+ // [2] SomeOperatorSpec.<init>()
+ // [3] OperatorSpecs.createSomeOperatorSpec()
+ // [4] MessageStreamImpl.someOperator()
+ // [5] User code that calls [4]
+ // we are interested in [5] here
+ StackTraceElement element = this.creationStackTrace[5];
+ return String.format("%s:%s", element.getFileName(), element.getLineNumber());
+ }
/**
* Get the name for this operator based on its opCode and opId.
* @return the name for this operator
*/
- default String getOpName() {
+ public final String getOpName() {
return String.format("%s-%s", getOpCode().name().toLowerCase(), getOpId());
}
-
}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java b/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java
index 66e2c58..ed5fc8f 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java
@@ -20,14 +20,11 @@
package org.apache.samza.operators.spec;
import org.apache.samza.config.Config;
-import org.apache.samza.operators.MessageStreamImpl;
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.PartialJoinFunction;
import org.apache.samza.operators.functions.SinkFunction;
-import org.apache.samza.operators.stream.OutputStreamInternal;
-import org.apache.samza.operators.windows.WindowPane;
import org.apache.samza.operators.windows.internal.WindowInternal;
import org.apache.samza.task.TaskContext;
@@ -46,14 +43,13 @@ public class OperatorSpecs {
* Creates a {@link StreamOperatorSpec} for {@link MapFunction}
*
* @param mapFn the map function
- * @param nextStream the output {@link MessageStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @param <M> type of input message
* @param <OM> type of output message
* @return the {@link StreamOperatorSpec}
*/
public static <M, OM> StreamOperatorSpec<M, OM> createMapOperatorSpec(
- MapFunction<? super M, ? extends OM> mapFn, MessageStreamImpl<OM> nextStream, int opId) {
+ MapFunction<? super M, ? extends OM> mapFn, int opId) {
return new StreamOperatorSpec<>(new FlatMapFunction<M, OM>() {
@Override
public Collection<OM> apply(M message) {
@@ -76,20 +72,19 @@ public class OperatorSpecs {
public void close() {
mapFn.close();
}
- }, nextStream, OperatorSpec.OpCode.MAP, opId);
+ }, OperatorSpec.OpCode.MAP, opId);
}
/**
* Creates a {@link StreamOperatorSpec} for {@link FilterFunction}
*
* @param filterFn the transformation function
- * @param nextStream the output {@link MessageStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @param <M> type of input message
* @return the {@link StreamOperatorSpec}
*/
public static <M> StreamOperatorSpec<M, M> createFilterOperatorSpec(
- FilterFunction<? super M> filterFn, MessageStreamImpl<M> nextStream, int opId) {
+ FilterFunction<? super M> filterFn, int opId) {
return new StreamOperatorSpec<>(new FlatMapFunction<M, M>() {
@Override
public Collection<M> apply(M message) {
@@ -111,23 +106,21 @@ public class OperatorSpecs {
public void close() {
filterFn.close();
}
-
- }, nextStream, OperatorSpec.OpCode.FILTER, opId);
+ }, OperatorSpec.OpCode.FILTER, opId);
}
/**
- * Creates a {@link StreamOperatorSpec}.
+ * Creates a {@link StreamOperatorSpec} for {@link FlatMapFunction}.
*
- * @param transformFn the transformation function
- * @param nextStream the output {@link MessageStreamImpl} to send messages to
+ * @param flatMapFn the transformation function
* @param opId the unique ID of the operator
* @param <M> type of input message
* @param <OM> type of output message
* @return the {@link StreamOperatorSpec}
*/
- public static <M, OM> StreamOperatorSpec<M, OM> createStreamOperatorSpec(
- FlatMapFunction<? super M, ? extends OM> transformFn, MessageStreamImpl<OM> nextStream, int opId) {
- return new StreamOperatorSpec<>((FlatMapFunction<M, OM>) transformFn, nextStream, OperatorSpec.OpCode.FLAT_MAP, opId);
+ public static <M, OM> StreamOperatorSpec<M, OM> createFlatMapOperatorSpec(
+ FlatMapFunction<? super M, ? extends OM> flatMapFn, int opId) {
+ return new StreamOperatorSpec<>((FlatMapFunction<M, OM>) flatMapFn, OperatorSpec.OpCode.FLAT_MAP, opId);
}
/**
@@ -139,91 +132,89 @@ public class OperatorSpecs {
* @return the {@link SinkOperatorSpec} for the sink operator
*/
public static <M> SinkOperatorSpec<M> createSinkOperatorSpec(SinkFunction<? super M> sinkFn, int opId) {
- return new SinkOperatorSpec<>((SinkFunction<M>) sinkFn, OperatorSpec.OpCode.SINK, opId);
+ return new SinkOperatorSpec<>((SinkFunction<M>) sinkFn, opId);
}
/**
- * Creates a {@link SinkOperatorSpec} for the sendTo operator.
+ * Creates a {@link OutputOperatorSpec} for the sendTo operator.
*
- * @param outputStream the {@link OutputStreamInternal} to send messages to
+ * @param outputStream the {@link OutputStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @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 OutputStreamInternal}
- * @return the {@link SinkOperatorSpec} for the sendTo operator
+ * @param <M> the type of message in the {@link OutputStreamImpl}
+ * @return the {@link OutputOperatorSpec} for the sendTo operator
*/
- public static <K, V, M> SinkOperatorSpec<M> createSendToOperatorSpec(
- OutputStreamInternal<K, V, M> outputStream, int opId) {
- return new SinkOperatorSpec<>(outputStream, OperatorSpec.OpCode.SEND_TO, opId);
+ public static <K, V, M> OutputOperatorSpec<M> createSendToOperatorSpec(
+ OutputStreamImpl<K, V, M> outputStream, int opId) {
+ return new OutputOperatorSpec<>(outputStream, OperatorSpec.OpCode.SEND_TO, opId);
}
/**
- * Creates a {@link SinkOperatorSpec} for the partitionBy operator.
+ * Creates a {@link OutputOperatorSpec} for the partitionBy operator.
*
- * @param outputStream the {@link OutputStreamInternal} to send messages to
+ * @param outputStream the {@link OutputStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @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 OutputStreamInternal}
- * @return the {@link SinkOperatorSpec} for the partitionBy operator
+ * @param <M> the type of message in the {@link OutputStreamImpl}
+ * @return the {@link OutputOperatorSpec} for the partitionBy operator
*/
- public static <K, V, M> SinkOperatorSpec<M> createPartitionByOperatorSpec(
- OutputStreamInternal<K, V, M> outputStream, int opId) {
- return new SinkOperatorSpec<>(outputStream, OperatorSpec.OpCode.PARTITION_BY, opId);
+ public static <K, V, M> OutputOperatorSpec<M> createPartitionByOperatorSpec(
+ OutputStreamImpl<K, V, M> outputStream, int opId) {
+ return new OutputOperatorSpec<>(outputStream, OperatorSpec.OpCode.PARTITION_BY, opId);
}
/**
* Creates a {@link WindowOperatorSpec}.
*
* @param window the description of the window.
- * @param nextStream the output {@link MessageStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @param <M> the type of input message
- * @param <WK> the type of key in the {@link WindowPane}
+ * @param <WK> the type of key in the window output
* @param <WV> the type of value in the window
* @return the {@link WindowOperatorSpec}
*/
public static <M, WK, WV> WindowOperatorSpec<M, WK, WV> createWindowOperatorSpec(
- WindowInternal<M, WK, WV> window, MessageStreamImpl<WindowPane<WK, WV>> nextStream, int opId) {
- return new WindowOperatorSpec<>(window, nextStream, opId);
+ WindowInternal<M, WK, WV> window, int opId) {
+ return new WindowOperatorSpec<>(window, opId);
}
/**
- * Creates a {@link PartialJoinOperatorSpec}.
+ * Creates a {@link JoinOperatorSpec}.
*
- * @param thisPartialJoinFn the partial join function for this message stream
- * @param otherPartialJoinFn the partial join function for the other message stream
+ * @param leftInputOpSpec the operator spec for the stream on the left side of the join
+ * @param rightInputOpSpec the operator spec for the stream on the right side of the join
+ * @param joinFn the user-defined join function to get join keys and results
* @param ttlMs the ttl in ms for retaining messages in each stream
- * @param nextStream the output {@link MessageStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @param <K> the type of join key
* @param <M> the type of input message
* @param <JM> the type of message in the other join stream
- * @param <RM> the type of message in the join output
- * @return the {@link PartialJoinOperatorSpec}
+ * @param <RM> the type of join result
+ * @return the {@link JoinOperatorSpec}
*/
- public static <K, M, JM, RM> PartialJoinOperatorSpec<K, M, JM, RM> createPartialJoinOperatorSpec(
- PartialJoinFunction<K, M, JM, RM> thisPartialJoinFn, PartialJoinFunction<K, JM, M, RM> otherPartialJoinFn,
- long ttlMs, MessageStreamImpl<RM> nextStream, int opId) {
- return new PartialJoinOperatorSpec<>(thisPartialJoinFn, otherPartialJoinFn, ttlMs, nextStream, opId);
+ public static <K, M, JM, RM> JoinOperatorSpec<K, M, JM, RM> createJoinOperatorSpec(
+ OperatorSpec<?, M> leftInputOpSpec, OperatorSpec<?, JM> rightInputOpSpec,
+ JoinFunction<K, M, JM, RM> joinFn, long ttlMs, int opId) {
+ return new JoinOperatorSpec<>(leftInputOpSpec, rightInputOpSpec, joinFn, ttlMs, opId);
}
/**
* Creates a {@link StreamOperatorSpec} with a merger function.
*
- * @param nextStream the output {@link MessageStreamImpl} to send messages to
* @param opId the unique ID of the operator
* @param <M> the type of input message
* @return the {@link StreamOperatorSpec} for the merge
*/
- public static <M> StreamOperatorSpec<M, M> createMergeOperatorSpec(MessageStreamImpl<M> nextStream, int opId) {
+ public static <M> StreamOperatorSpec<M, M> createMergeOperatorSpec(int opId) {
return new StreamOperatorSpec<>(message ->
new ArrayList<M>() {
{
this.add(message);
}
},
- nextStream, OperatorSpec.OpCode.MERGE, opId);
+ OperatorSpec.OpCode.MERGE, opId);
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/OutputOperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/OutputOperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/OutputOperatorSpec.java
new file mode 100644
index 0000000..e6767ec
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/OutputOperatorSpec.java
@@ -0,0 +1,55 @@
+/*
+ * 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.spec;
+
+
+/**
+ * The spec for an operator that outputs a {@link org.apache.samza.operators.MessageStream} to a
+ * {@link org.apache.samza.system.SystemStream}.
+ * <p>
+ * This is a terminal operator and does not allow further operator chaining.
+ *
+ * @param <M> the type of input message
+ */
+public class OutputOperatorSpec<M> extends OperatorSpec<M, Void> {
+
+ private OutputStreamImpl<?, ?, M> outputStream;
+
+
+ /**
+ * Constructs an {@link OutputOperatorSpec} to send messages to the provided {@code outStream}
+ *
+ * @param outputStream the {@link OutputStreamImpl} to send messages to
+ * @param opCode the specific {@link OpCode} for this {@link SinkOperatorSpec}.
+ * It could be {@link OpCode#SEND_TO}, or {@link OpCode#PARTITION_BY}
+ * @param opId the unique ID of this {@link SinkOperatorSpec} in the graph
+ */
+ OutputOperatorSpec(OutputStreamImpl<?, ?, M> outputStream, OperatorSpec.OpCode opCode, int opId) {
+ super(opCode, opId);
+ this.outputStream = outputStream;
+ }
+
+ /**
+ * The {@link OutputStreamImpl} that this operator is sending its output to.
+ * @return the {@link OutputStreamImpl} for this operator if any, else null.
+ */
+ public OutputStreamImpl<?, ?, M> getOutputStream() {
+ return this.outputStream;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/OutputStreamImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/OutputStreamImpl.java b/samza-core/src/main/java/org/apache/samza/operators/spec/OutputStreamImpl.java
new file mode 100644
index 0000000..5506378
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/OutputStreamImpl.java
@@ -0,0 +1,50 @@
+/*
+ * 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.spec;
+
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.system.StreamSpec;
+
+import java.util.function.Function;
+
+public class OutputStreamImpl<K, V, M> implements OutputStream<K, V, M> {
+
+ private final StreamSpec streamSpec;
+ private final Function<M, K> keyExtractor;
+ private final Function<M, V> msgExtractor;
+
+ public OutputStreamImpl(StreamSpec streamSpec,
+ Function<M, K> keyExtractor, Function<M, V> msgExtractor) {
+ this.streamSpec = streamSpec;
+ this.keyExtractor = keyExtractor;
+ this.msgExtractor = msgExtractor;
+ }
+
+ public StreamSpec getStreamSpec() {
+ return streamSpec;
+ }
+
+ public Function<M, K> getKeyExtractor() {
+ return keyExtractor;
+ }
+
+ public Function<M, V> getMsgExtractor() {
+ return msgExtractor;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/PartialJoinOperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/PartialJoinOperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/PartialJoinOperatorSpec.java
deleted file mode 100644
index 92b4170..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/spec/PartialJoinOperatorSpec.java
+++ /dev/null
@@ -1,97 +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.spec;
-
-import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.functions.PartialJoinFunction;
-import org.apache.samza.operators.util.OperatorJsonUtils;
-
-
-/**
- * Spec for the partial join operator that takes messages from one input stream, joins with buffered
- * messages from another stream, and produces join results to an output {@link MessageStreamImpl}.
- *
- * @param <K> the type of join key
- * @param <M> the type of input message
- * @param <JM> the type of message in the other join stream
- * @param <RM> the type of message in the join output stream
- */
-public class PartialJoinOperatorSpec<K, M, JM, RM> implements OperatorSpec<RM> {
-
- private final PartialJoinFunction<K, M, JM, RM> thisPartialJoinFn;
- private final PartialJoinFunction<K, JM, M, RM> otherPartialJoinFn;
- private final long ttlMs;
- private final MessageStreamImpl<RM> nextStream;
- private final int opId;
- private final String sourceLocation;
-
- /**
- * Default constructor for a {@link PartialJoinOperatorSpec}.
- *
- * @param thisPartialJoinFn partial join function that provides state and the join logic for input messages of
- * type {@code M} in this stream
- * @param otherPartialJoinFn partial join function that provides state for input messages of type {@code JM}
- * in the other stream
- * @param ttlMs the ttl in ms for retaining messages in each stream
- * @param nextStream the output {@link MessageStreamImpl} containing the messages produced from this operator
- * @param opId the unique ID for this operator
- */
- PartialJoinOperatorSpec(PartialJoinFunction<K, M, JM, RM> thisPartialJoinFn,
- PartialJoinFunction<K, JM, M, RM> otherPartialJoinFn, long ttlMs,
- MessageStreamImpl<RM> nextStream, int opId) {
- this.thisPartialJoinFn = thisPartialJoinFn;
- this.otherPartialJoinFn = otherPartialJoinFn;
- this.ttlMs = ttlMs;
- this.nextStream = nextStream;
- this.opId = opId;
- this.sourceLocation = OperatorJsonUtils.getSourceLocation();
- }
-
- @Override
- public MessageStreamImpl<RM> getNextStream() {
- return this.nextStream;
- }
-
- public PartialJoinFunction<K, M, JM, RM> getThisPartialJoinFn() {
- return this.thisPartialJoinFn;
- }
-
- public PartialJoinFunction<K, JM, M, RM> getOtherPartialJoinFn() {
- return this.otherPartialJoinFn;
- }
-
- public long getTtlMs() {
- return ttlMs;
- }
-
- @Override
- public OperatorSpec.OpCode getOpCode() {
- return OpCode.JOIN;
- }
-
- @Override
- public int getOpId() {
- return this.opId;
- }
-
- @Override
- public String getSourceLocation() {
- return sourceLocation;
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/SinkOperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/SinkOperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/SinkOperatorSpec.java
index afdd6b9..2b55d95 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/spec/SinkOperatorSpec.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/SinkOperatorSpec.java
@@ -18,29 +18,19 @@
*/
package org.apache.samza.operators.spec;
-import org.apache.samza.operators.MessageStreamImpl;
import org.apache.samza.operators.functions.SinkFunction;
-import org.apache.samza.operators.stream.OutputStreamInternal;
-import org.apache.samza.operators.util.OperatorJsonUtils;
-import org.apache.samza.system.OutgoingMessageEnvelope;
-import org.apache.samza.system.SystemStream;
-import org.apache.samza.task.MessageCollector;
-import org.apache.samza.task.TaskCoordinator;
/**
- * The spec for an operator that outputs a {@link MessageStreamImpl} to an external system.
+ * The spec for an operator that outputs a stream to an arbitrary external system.
+ * <p>
* This is a terminal operator and does not allow further operator chaining.
*
* @param <M> the type of input message
*/
-public class SinkOperatorSpec<M> implements OperatorSpec {
+public class SinkOperatorSpec<M> extends OperatorSpec<M, Void> {
private final SinkFunction<M> sinkFn;
- private OutputStreamInternal<?, ?, M> outputStream; // may be null
- private final OperatorSpec.OpCode opCode;
- private final int opId;
- private final String sourceLocation;
/**
* Constructs a {@link SinkOperatorSpec} with a user defined {@link SinkFunction}.
@@ -48,79 +38,14 @@ public class SinkOperatorSpec<M> implements OperatorSpec {
* @param sinkFn a user defined {@link SinkFunction} that will be called with the output message,
* the output {@link org.apache.samza.task.MessageCollector} and the
* {@link org.apache.samza.task.TaskCoordinator}.
- * @param opCode the specific {@link OpCode} for this {@link SinkOperatorSpec}.
- * It could be {@link OpCode#SINK}, {@link OpCode#SEND_TO}, or {@link OpCode#PARTITION_BY}.
* @param opId the unique ID of this {@link OperatorSpec} in the graph
*/
- SinkOperatorSpec(SinkFunction<M> sinkFn, OperatorSpec.OpCode opCode, int opId) {
+ SinkOperatorSpec(SinkFunction<M> sinkFn, int opId) {
+ super(OpCode.SINK, opId);
this.sinkFn = sinkFn;
- this.opCode = opCode;
- this.opId = opId;
- this.sourceLocation = OperatorJsonUtils.getSourceLocation();
- }
-
- /**
- * Constructs a {@link SinkOperatorSpec} to send messages to the provided {@code outStream}
- * @param outputStream the {@link OutputStreamInternal} to send messages to
- * @param opCode the specific {@link OpCode} for this {@link SinkOperatorSpec}.
- * It could be {@link OpCode#SINK}, {@link OpCode#SEND_TO}, or {@link OpCode#PARTITION_BY}
- * @param opId the unique ID of this {@link SinkOperatorSpec} in the graph
- */
- SinkOperatorSpec(OutputStreamInternal<?, ?, M> outputStream, OperatorSpec.OpCode opCode, int opId) {
- this(createSinkFn(outputStream), opCode, opId);
- this.outputStream = outputStream;
- }
-
- /**
- * This is a terminal operator and doesn't allow further operator chaining.
- * @return null
- */
- @Override
- public MessageStreamImpl<M> getNextStream() {
- return null;
- }
-
- /**
- * The {@link OutputStreamInternal} that this operator is sending its output to.
- * @return the {@link OutputStreamInternal} for this operator if any, else null.
- */
- public OutputStreamInternal<?, ?, M> getOutputStream() {
- return this.outputStream;
}
public SinkFunction<M> getSinkFn() {
return this.sinkFn;
}
-
- @Override
- public OperatorSpec.OpCode getOpCode() {
- return this.opCode;
- }
-
- @Override
- public int getOpId() {
- return this.opId;
- }
-
- @Override
- public String getSourceLocation() {
- return sourceLocation;
- }
-
- /**
- * Creates a {@link SinkFunction} to send messages to the provided {@code output}.
- * @param outputStream the {@link OutputStreamInternal} to send messages to
- * @param <M> the type of input message
- * @return a {@link SinkFunction} that sends messages to the provided {@code output}
- */
- private static <M> SinkFunction<M> createSinkFn(OutputStreamInternal<?, ?, M> outputStream) {
- return (M message, MessageCollector mc, TaskCoordinator tc) -> {
- // TODO: SAMZA-1148 - need to find a way to directly pass in the serde class names
- SystemStream systemStream = new SystemStream(outputStream.getStreamSpec().getSystemName(),
- outputStream.getStreamSpec().getPhysicalName());
- Object key = outputStream.getKeyExtractor().apply(message);
- Object msg = outputStream.getMsgExtractor().apply(message);
- mc.send(new OutgoingMessageEnvelope(systemStream, key, msg));
- };
- }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/StreamOperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/StreamOperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/StreamOperatorSpec.java
index c53efae..1f2f683 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/spec/StreamOperatorSpec.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/StreamOperatorSpec.java
@@ -18,63 +18,32 @@
*/
package org.apache.samza.operators.spec;
-import org.apache.samza.operators.MessageStreamImpl;
import org.apache.samza.operators.functions.FlatMapFunction;
-import org.apache.samza.operators.util.OperatorJsonUtils;
/**
- * The spec for a linear stream operator that outputs 0 or more messages for each input message.
+ * The spec for a simple stream operator that outputs 0 or more messages for each input message.
*
* @param <M> the type of input message
* @param <OM> the type of output message
*/
-public class StreamOperatorSpec<M, OM> implements OperatorSpec<OM> {
+public class StreamOperatorSpec<M, OM> extends OperatorSpec<M, OM> {
private final FlatMapFunction<M, OM> transformFn;
- private final MessageStreamImpl<OM> nextStream;
- private final OperatorSpec.OpCode opCode;
- private final int opId;
- private final String sourceLocation;
/**
- * Constructor for a {@link StreamOperatorSpec} that accepts an output {@link MessageStreamImpl}.
+ * Constructor for a {@link StreamOperatorSpec}.
*
* @param transformFn the transformation function
- * @param nextStream the output {@link MessageStreamImpl} containing the messages produced from this operator
* @param opCode the {@link OpCode} for this {@link StreamOperatorSpec}
- * @param opId the unique id for this {@link StreamOperatorSpec} in a {@link org.apache.samza.operators.StreamGraph}
+ * @param opId the unique ID for this {@link StreamOperatorSpec}
*/
- StreamOperatorSpec(FlatMapFunction<M, OM> transformFn, MessageStreamImpl<OM> nextStream,
- OperatorSpec.OpCode opCode, int opId) {
+ StreamOperatorSpec(FlatMapFunction<M, OM> transformFn, OperatorSpec.OpCode opCode, int opId) {
+ super(opCode, opId);
this.transformFn = transformFn;
- this.nextStream = nextStream;
- this.opCode = opCode;
- this.opId = opId;
- this.sourceLocation = OperatorJsonUtils.getSourceLocation();
- }
-
- @Override
- public MessageStreamImpl<OM> getNextStream() {
- return this.nextStream;
}
public FlatMapFunction<M, OM> getTransformFn() {
return this.transformFn;
}
-
- @Override
- public OperatorSpec.OpCode getOpCode() {
- return this.opCode;
- }
-
- @Override
- public int getOpId() {
- return this.opId;
- }
-
- @Override
- public String getSourceLocation() {
- return sourceLocation;
- }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/spec/WindowOperatorSpec.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/spec/WindowOperatorSpec.java b/samza-core/src/main/java/org/apache/samza/operators/spec/WindowOperatorSpec.java
index 3c2be0a..0937499 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/spec/WindowOperatorSpec.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/spec/WindowOperatorSpec.java
@@ -19,13 +19,11 @@
package org.apache.samza.operators.spec;
-import org.apache.samza.operators.MessageStreamImpl;
import org.apache.samza.operators.triggers.AnyTrigger;
import org.apache.samza.operators.triggers.RepeatingTrigger;
import org.apache.samza.operators.triggers.TimeBasedTrigger;
import org.apache.samza.operators.triggers.Trigger;
import org.apache.samza.operators.util.MathUtils;
-import org.apache.samza.operators.util.OperatorJsonUtils;
import org.apache.samza.operators.windows.WindowPane;
import org.apache.samza.operators.windows.internal.WindowInternal;
import org.slf4j.Logger;
@@ -37,58 +35,32 @@ import java.util.stream.Collectors;
/**
- * Default window operator spec object
+ * The spec for an operator that groups messages into finite windows for processing
*
* @param <M> the type of input message to the window
* @param <WK> the type of key of the window
* @param <WV> the type of aggregated value in the window output {@link WindowPane}
*/
-public class WindowOperatorSpec<M, WK, WV> implements OperatorSpec<WindowPane<WK, WV>> {
+public class WindowOperatorSpec<M, WK, WV> extends OperatorSpec<M, WindowPane<WK, WV>> {
private static final Logger LOG = LoggerFactory.getLogger(WindowOperatorSpec.class);
private final WindowInternal<M, WK, WV> window;
- private final MessageStreamImpl<WindowPane<WK, WV>> nextStream;
- private final int opId;
- private final String sourceLocation;
/**
* Constructor for {@link WindowOperatorSpec}.
*
* @param window the window function
- * @param nextStream the output {@link MessageStreamImpl} containing the messages produced from this operator
* @param opId auto-generated unique ID of this operator
*/
- WindowOperatorSpec(WindowInternal<M, WK, WV> window, MessageStreamImpl<WindowPane<WK, WV>> nextStream, int opId) {
- this.nextStream = nextStream;
+ WindowOperatorSpec(WindowInternal<M, WK, WV> window, int opId) {
+ super(OpCode.WINDOW, opId);
this.window = window;
- this.opId = opId;
- this.sourceLocation = OperatorJsonUtils.getSourceLocation();
- }
-
- @Override
- public MessageStreamImpl<WindowPane<WK, WV>> getNextStream() {
- return this.nextStream;
}
public WindowInternal<M, WK, WV> getWindow() {
return window;
}
- @Override
- public OpCode getOpCode() {
- return OpCode.WINDOW;
- }
-
- @Override
- public int getOpId() {
- return this.opId;
- }
-
- @Override
- public String getSourceLocation() {
- return sourceLocation;
- }
-
/**
* Get the default triggering interval for this {@link WindowOperatorSpec}
*
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternal.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternal.java b/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternal.java
deleted file mode 100644
index e67b326..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternal.java
+++ /dev/null
@@ -1,39 +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.stream;
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.MessageStream;
-import org.apache.samza.system.StreamSpec;
-
-import java.util.function.BiFunction;
-
-/**
- * Internal representation of an input stream.
- *
- * @param <M> the type of messages in the input stream
- */
-@InterfaceStability.Unstable
-public interface InputStreamInternal<K, V, M> extends MessageStream<M> {
-
- StreamSpec getStreamSpec();
-
- BiFunction<K, V, M> getMsgBuilder();
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternalImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternalImpl.java b/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternalImpl.java
deleted file mode 100644
index c4337d0..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/stream/InputStreamInternalImpl.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.stream;
-
-import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.StreamGraphImpl;
-import org.apache.samza.system.StreamSpec;
-
-import java.util.function.BiFunction;
-
-public class InputStreamInternalImpl<K, V, M> extends MessageStreamImpl<M> implements InputStreamInternal<K, V, M> {
-
- private final StreamSpec streamSpec;
- private final BiFunction<K, V, M> msgBuilder;
-
- public InputStreamInternalImpl(StreamGraphImpl graph, StreamSpec streamSpec, BiFunction<K, V, M> msgBuilder) {
- super(graph);
- this.streamSpec = streamSpec;
- this.msgBuilder = msgBuilder;
- }
-
- public StreamSpec getStreamSpec() {
- return this.streamSpec;
- }
-
- public BiFunction<K, V, M> getMsgBuilder() {
- return this.msgBuilder;
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateMessageStreamImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateMessageStreamImpl.java b/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateMessageStreamImpl.java
new file mode 100644
index 0000000..f0bb1dc
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateMessageStreamImpl.java
@@ -0,0 +1,58 @@
+/*
+ * 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.stream;
+
+import org.apache.samza.operators.MessageStreamImpl;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.operators.StreamGraphImpl;
+import org.apache.samza.operators.spec.InputOperatorSpec;
+import org.apache.samza.operators.spec.OutputStreamImpl;
+import org.apache.samza.system.StreamSpec;
+
+/**
+ * An intermediate stream is both an input and an output stream (e.g. a repartitioned stream).
+ * <p>
+ * This implementation accepts a pair of {@link InputOperatorSpec} and {@link OutputStreamImpl} associated
+ * with the same logical {@code streamId}. It provides access to its {@link OutputStreamImpl} for
+ * {@link MessageStreamImpl#partitionBy} to send messages out to. It's also a {@link MessageStreamImpl} with
+ * {@link InputOperatorSpec} as its operator spec, so that further operations can be chained on the
+ * {@link InputOperatorSpec}.
+ *
+ * @param <K> the type of key in the outgoing/incoming message
+ * @param <V> the type of message in the outgoing/incoming message
+ * @param <M> the type of message in the output {@link MessageStreamImpl}
+ */
+public class IntermediateMessageStreamImpl<K, V, M> extends MessageStreamImpl<M> implements OutputStream<K, V, M> {
+
+ private final OutputStreamImpl<K, V, M> outputStream;
+
+ public IntermediateMessageStreamImpl(StreamGraphImpl graph, InputOperatorSpec<K, V, M> inputOperatorSpec,
+ OutputStreamImpl<K, V, M> outputStream) {
+ super(graph, inputOperatorSpec);
+ this.outputStream = outputStream;
+ }
+
+ public StreamSpec getStreamSpec() {
+ return this.outputStream.getStreamSpec();
+ }
+
+ public OutputStreamImpl<K, V, M> getOutputStream() {
+ return this.outputStream;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateStreamInternalImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateStreamInternalImpl.java b/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateStreamInternalImpl.java
deleted file mode 100644
index 8f45f7a..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateStreamInternalImpl.java
+++ /dev/null
@@ -1,61 +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.stream;
-
-import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.StreamGraphImpl;
-import org.apache.samza.system.StreamSpec;
-
-import java.util.function.BiFunction;
-import java.util.function.Function;
-
-public class IntermediateStreamInternalImpl<K, V, M> extends MessageStreamImpl<M>
- implements InputStreamInternal<K, V, M>, OutputStreamInternal<K, V, M> {
-
- private final StreamSpec streamSpec;
- private final Function<M, K> keyExtractor;
- private final Function<M, V> msgExtractor;
- private final BiFunction<K, V, M> msgBuilder;
-
- public IntermediateStreamInternalImpl(StreamGraphImpl graph, StreamSpec streamSpec, Function<M, K> keyExtractor,
- Function<M, V> msgExtractor, BiFunction<K, V, M> msgBuilder) {
- super(graph);
- this.streamSpec = streamSpec;
- this.keyExtractor = keyExtractor;
- this.msgExtractor = msgExtractor;
- this.msgBuilder = msgBuilder;
- }
-
- public StreamSpec getStreamSpec() {
- return this.streamSpec;
- }
-
- public Function<M, K> getKeyExtractor() {
- return this.keyExtractor;
- }
-
- public Function<M, V> getMsgExtractor() {
- return this.msgExtractor;
- }
-
- @Override
- public BiFunction<K, V, M> getMsgBuilder() {
- return this.msgBuilder;
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternal.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternal.java b/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternal.java
deleted file mode 100644
index 48ce641..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternal.java
+++ /dev/null
@@ -1,43 +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.stream;
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.OutputStream;
-import org.apache.samza.system.StreamSpec;
-
-import java.util.function.Function;
-
-
-/**
- * Internal representation of an output stream.
- *
- * @param <M> the type of messages in the output stream
- */
-@InterfaceStability.Unstable
-public interface OutputStreamInternal<K, V, M> extends OutputStream<K, V, M> {
-
- StreamSpec getStreamSpec();
-
- Function<M, K> getKeyExtractor();
-
- Function<M, V> getMsgExtractor();
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternalImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternalImpl.java b/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternalImpl.java
deleted file mode 100644
index a2d0cca..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/stream/OutputStreamInternalImpl.java
+++ /dev/null
@@ -1,52 +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.stream;
-
-import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.StreamGraphImpl;
-import org.apache.samza.system.StreamSpec;
-
-import java.util.function.Function;
-
-public class OutputStreamInternalImpl<K, V, M> extends MessageStreamImpl<M> implements OutputStreamInternal<K, V, M> {
-
- private final StreamSpec streamSpec;
- private final Function<M, K> keyExtractor;
- private final Function<M, V> msgExtractor;
-
- public OutputStreamInternalImpl(StreamGraphImpl graph, StreamSpec streamSpec,
- Function<M, K> keyExtractor, Function<M, V> msgExtractor) {
- super(graph);
- this.streamSpec = streamSpec;
- this.keyExtractor = keyExtractor;
- this.msgExtractor = msgExtractor;
- }
-
- public StreamSpec getStreamSpec() {
- return this.streamSpec;
- }
-
- public Function<M, K> getKeyExtractor() {
- return this.keyExtractor;
- }
-
- public Function<M, V> getMsgExtractor() {
- return this.msgExtractor;
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/operators/util/OperatorJsonUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/operators/util/OperatorJsonUtils.java b/samza-core/src/main/java/org/apache/samza/operators/util/OperatorJsonUtils.java
deleted file mode 100644
index b971607..0000000
--- a/samza-core/src/main/java/org/apache/samza/operators/util/OperatorJsonUtils.java
+++ /dev/null
@@ -1,89 +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.util;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.stream.Collectors;
-import org.apache.samza.operators.spec.OperatorSpec;
-import org.apache.samza.operators.spec.PartialJoinOperatorSpec;
-import org.apache.samza.operators.spec.SinkOperatorSpec;
-import org.apache.samza.operators.stream.OutputStreamInternal;
-
-public class OperatorJsonUtils {
- private static final String OP_CODE = "opCode";
- private static final String OP_ID = "opId";
- private static final String SOURCE_LOCATION = "sourceLocation";
- private static final String NEXT_OPERATOR_IDS = "nextOperatorIds";
- private static final String OUTPUT_STREAM_ID = "outputStreamId";
- private static final String TTL_MS = "ttlMs";
-
- /**
- * Format the operator properties into a map
- * @param spec a {@link OperatorSpec} instance
- * @return map of the operator properties
- */
- public static Map<String, Object> operatorToMap(OperatorSpec spec) {
- Map<String, Object> map = new HashMap<>();
- map.put(OP_CODE, spec.getOpCode().name());
- map.put(OP_ID, spec.getOpId());
- map.put(SOURCE_LOCATION, spec.getSourceLocation());
-
- if (spec.getNextStream() != null) {
- Collection<OperatorSpec> nextOperators = spec.getNextStream().getRegisteredOperatorSpecs();
- map.put(NEXT_OPERATOR_IDS, nextOperators.stream().map(OperatorSpec::getOpId).collect(Collectors.toSet()));
- } else {
- map.put(NEXT_OPERATOR_IDS, Collections.emptySet());
- }
-
- if (spec instanceof SinkOperatorSpec) {
- OutputStreamInternal outputStream = ((SinkOperatorSpec) spec).getOutputStream();
- if (outputStream != null) {
- map.put(OUTPUT_STREAM_ID, outputStream.getStreamSpec().getId());
- }
- }
-
- if (spec instanceof PartialJoinOperatorSpec) {
- map.put(TTL_MS, ((PartialJoinOperatorSpec) spec).getTtlMs());
- }
-
- return map;
- }
-
- /**
- * Return the location of source code that creates the operator.
- * This function is invoked in the constructor of each operator.
- * @return formatted source location including file and line number
- */
- public static String getSourceLocation() {
- // The stack trace looks like:
- // [0] Thread.getStackTrace()
- // [1] OperatorJsonUtils.getSourceLocation()
- // [2] SomeOperator.<init>()
- // [3] OperatorSpecs.createSomeOperator()
- // [4] MessageStreamImpl.someOperator()
- // [5] User code that calls [2]
- // we are only interested in [5] here
- StackTraceElement location = Thread.currentThread().getStackTrace()[5];
- return String.format("%s:%s", location.getFileName(), location.getLineNumber());
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java b/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
index 50ae775..a77ef3b 100644
--- a/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
+++ b/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
@@ -18,22 +18,19 @@
*/
package org.apache.samza.task;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.ContextManager;
import org.apache.samza.operators.StreamGraphImpl;
+import org.apache.samza.operators.impl.InputOperatorImpl;
import org.apache.samza.operators.impl.OperatorImplGraph;
-import org.apache.samza.operators.impl.RootOperatorImpl;
-import org.apache.samza.operators.stream.InputStreamInternal;
import org.apache.samza.runtime.ApplicationRunner;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.SystemStream;
import org.apache.samza.util.Clock;
import org.apache.samza.util.SystemClock;
-import java.util.HashMap;
-import java.util.Map;
-
/**
* A {@link StreamTask} implementation that brings all the operator API implementation components together and
@@ -47,7 +44,6 @@ public final class StreamOperatorTask implements StreamTask, InitableTask, Windo
private OperatorImplGraph operatorImplGraph;
private ContextManager contextManager;
- private Map<SystemStream, InputStreamInternal> inputSystemStreamToInputStream;
/**
* Constructs an adaptor task to run the user-implemented {@link StreamApplication}.
@@ -70,11 +66,8 @@ public final class StreamOperatorTask implements StreamTask, InitableTask, Windo
* <p>
* Implementation: Initializes the user-implemented {@link StreamApplication}. The {@link StreamApplication} sets
* the input and output streams and the task-wide context manager using the {@link StreamGraphImpl} APIs,
- * and the logical transforms using the {@link org.apache.samza.operators.MessageStream} APIs.
- *<p>
- * It then uses the {@link StreamGraphImpl} to create the {@link OperatorImplGraph} corresponding to the logical
- * DAG. It also saves the mapping between input {@link SystemStream}s and their corresponding
- * {@link InputStreamInternal}s for delivering incoming messages to the appropriate sub-DAG.
+ * and the logical transforms using the {@link org.apache.samza.operators.MessageStream} APIs. It then uses
+ * the {@link StreamGraphImpl} to create the {@link OperatorImplGraph} corresponding to the logical DAG.
*
* @param config allows accessing of fields in the configuration files that this StreamTask is specified in
* @param context allows initializing and accessing contextual data of this StreamTask
@@ -93,20 +86,11 @@ public final class StreamOperatorTask implements StreamTask, InitableTask, Windo
}
// create the operator impl DAG corresponding to the logical operator spec DAG
- OperatorImplGraph operatorImplGraph = new OperatorImplGraph(clock);
- operatorImplGraph.init(streamGraph, config, context);
- this.operatorImplGraph = operatorImplGraph;
-
- // TODO: SAMZA-1118 - Remove mapping after SystemConsumer starts returning logical streamId with incoming messages
- inputSystemStreamToInputStream = new HashMap<>();
- streamGraph.getInputStreams().forEach((streamSpec, inputStream)-> {
- SystemStream systemStream = new SystemStream(streamSpec.getSystemName(), streamSpec.getPhysicalName());
- inputSystemStreamToInputStream.put(systemStream, inputStream);
- });
+ this.operatorImplGraph = new OperatorImplGraph(streamGraph, config, context, clock);
}
/**
- * Passes the incoming message envelopes along to the {@link org.apache.samza.operators.impl.RootOperatorImpl} node
+ * Passes the incoming message envelopes along to the {@link InputOperatorImpl} node
* for the input {@link SystemStream}.
* <p>
* From then on, each {@link org.apache.samza.operators.impl.OperatorImpl} propagates its transformed output to
@@ -119,20 +103,16 @@ public final class StreamOperatorTask implements StreamTask, InitableTask, Windo
@Override
public final void process(IncomingMessageEnvelope ime, MessageCollector collector, TaskCoordinator coordinator) {
SystemStream systemStream = ime.getSystemStreamPartition().getSystemStream();
- InputStreamInternal inputStream = inputSystemStreamToInputStream.get(systemStream);
- RootOperatorImpl rootOperatorImpl = operatorImplGraph.getRootOperator(systemStream);
- if (rootOperatorImpl != null) {
- // TODO: SAMZA-1148 - Cast to appropriate input (key, msg) types based on the serde
- // before applying the msgBuilder.
- Object message = inputStream.getMsgBuilder().apply(ime.getKey(), ime.getMessage());
- rootOperatorImpl.onMessage(message, collector, coordinator);
+ InputOperatorImpl inputOpImpl = operatorImplGraph.getInputOperator(systemStream);
+ if (inputOpImpl != null) {
+ inputOpImpl.onMessage(Pair.of(ime.getKey(), ime.getMessage()), collector, coordinator);
}
}
@Override
public final void window(MessageCollector collector, TaskCoordinator coordinator) {
- operatorImplGraph.getAllRootOperators()
- .forEach(rootOperator -> rootOperator.onTimer(collector, coordinator));
+ operatorImplGraph.getAllInputOperators()
+ .forEach(inputOperator -> inputOperator.onTimer(collector, coordinator));
}
@Override
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java b/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
index daa223a..2c8f682 100644
--- a/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
+++ b/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
@@ -224,12 +224,12 @@ public class TestExecutionPlanner {
when(runner.getStreamSpec("output2")).thenReturn(output2);
// intermediate streams used in tests
- when(runner.getStreamSpec("test-app-1-partition_by-0"))
- .thenReturn(new StreamSpec("test-app-1-partition_by-0", "test-app-1-partition_by-0", "default-system"));
when(runner.getStreamSpec("test-app-1-partition_by-1"))
.thenReturn(new StreamSpec("test-app-1-partition_by-1", "test-app-1-partition_by-1", "default-system"));
- when(runner.getStreamSpec("test-app-1-partition_by-4"))
- .thenReturn(new StreamSpec("test-app-1-partition_by-4", "test-app-1-partition_by-4", "default-system"));
+ when(runner.getStreamSpec("test-app-1-partition_by-3"))
+ .thenReturn(new StreamSpec("test-app-1-partition_by-3", "test-app-1-partition_by-3", "default-system"));
+ when(runner.getStreamSpec("test-app-1-partition_by-8"))
+ .thenReturn(new StreamSpec("test-app-1-partition_by-8", "test-app-1-partition_by-8", "default-system"));
}
@Test
@@ -272,7 +272,7 @@ public class TestExecutionPlanner {
// the partitions should be the same as input1
jobGraph.getIntermediateStreams().forEach(edge -> {
- assertTrue(edge.getPartitionCount() == 64);
+ assertEquals(64, edge.getPartitionCount());
});
}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java b/samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java
index e53cd42..4bda86b 100644
--- a/samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java
+++ b/samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java
@@ -39,7 +39,7 @@ import org.codehaus.jackson.map.ObjectMapper;
import org.junit.Test;
import static org.apache.samza.execution.TestExecutionPlanner.createSystemAdmin;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -50,13 +50,16 @@ public class TestJobGraphJsonGenerator {
public void test() throws Exception {
/**
- * the graph looks like the following. number of partitions in parentheses. quotes indicate expected value.
+ * the graph looks like the following.
+ * number in parentheses () indicates number of stream partitions.
+ * number in parentheses in quotes ("") indicates expected partition count.
+ * number in square brackets [] indicates operator ID.
*
- * input1 (64) -> map -> join -> output1 (8)
- * |
- * input2 (16) -> partitionBy ("64") -> filter -|
- * |
- * input3 (32) -> filter -> partitionBy ("64") -> map -> join -> output2 (16)
+ * input3 (32) -> filter [7] -> partitionBy [8] ("64") -> map [10] -> join [14] -> sendTo(output2) [15] (16)
+ * |
+ * input2 (16) -> partitionBy [3] ("64") -> filter [5] -| -> sink [13]
+ * |
+ * input1 (64) -> map [1] -> join [11] -> sendTo(output1) [12] (8)
*
*/
@@ -80,12 +83,10 @@ public class TestJobGraphJsonGenerator {
when(runner.getStreamSpec("output2")).thenReturn(output2);
// intermediate streams used in tests
- when(runner.getStreamSpec("test-app-1-partition_by-0"))
- .thenReturn(new StreamSpec("test-app-1-partition_by-0", "test-app-1-partition_by-0", "default-system"));
- when(runner.getStreamSpec("test-app-1-partition_by-1"))
- .thenReturn(new StreamSpec("test-app-1-partition_by-1", "test-app-1-partition_by-1", "default-system"));
- when(runner.getStreamSpec("test-app-1-partition_by-4"))
- .thenReturn(new StreamSpec("test-app-1-partition_by-4", "test-app-1-partition_by-4", "default-system"));
+ when(runner.getStreamSpec("test-app-1-partition_by-3"))
+ .thenReturn(new StreamSpec("test-app-1-partition_by-3", "test-app-1-partition_by-3", "default-system"));
+ when(runner.getStreamSpec("test-app-1-partition_by-8"))
+ .thenReturn(new StreamSpec("test-app-1-partition_by-8", "test-app-1-partition_by-8", "default-system"));
// set up external partition count
Map<String, Integer> system1Map = new HashMap<>();
@@ -124,10 +125,10 @@ public class TestJobGraphJsonGenerator {
// deserialize
ObjectMapper mapper = new ObjectMapper();
JobGraphJsonGenerator.JobGraphJson nodes = mapper.readValue(json, JobGraphJsonGenerator.JobGraphJson.class);
- assertTrue(nodes.jobs.get(0).operatorGraph.inputStreams.size() == 5);
- assertTrue(nodes.jobs.get(0).operatorGraph.operators.size() == 13);
- assertTrue(nodes.sourceStreams.size() == 3);
- assertTrue(nodes.sinkStreams.size() == 2);
- assertTrue(nodes.intermediateStreams.size() == 2);
+ assertEquals(5, nodes.jobs.get(0).operatorGraph.inputStreams.size());
+ assertEquals(11, nodes.jobs.get(0).operatorGraph.operators.size());
+ assertEquals(3, nodes.sourceStreams.size());
+ assertEquals(2, nodes.sinkStreams.size());
+ assertEquals(2, nodes.intermediateStreams.size());
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/c1c4289c/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java b/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
index 39745bf..0c41fb8 100644
--- a/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
+++ b/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
@@ -74,18 +74,18 @@ public class TestJoinOperator {
public void testJoinFnInitAndClose() throws Exception {
TestJoinFunction joinFn = new TestJoinFunction();
StreamOperatorTask sot = createStreamOperatorTask(new SystemClock(), new TestJoinStreamApplication(joinFn));
- assertEquals(joinFn.getNumInitCalls(), 1);
+ assertEquals(1, joinFn.getNumInitCalls());
MessageCollector messageCollector = mock(MessageCollector.class);
// push messages to first stream
numbers.forEach(n -> sot.process(new FirstStreamIME(n, n), messageCollector, taskCoordinator));
// close should not be called till now
- assertEquals(joinFn.getNumCloseCalls(), 0);
+ assertEquals(0, joinFn.getNumCloseCalls());
sot.close();
// close should be called from sot.close()
- assertEquals(joinFn.getNumCloseCalls(), 1);
+ assertEquals(1, joinFn.getNumCloseCalls());
}
@Test