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/05/15 21:38:57 UTC
[1/5] samza git commit: SAMZA-1286: Close zk connection in
ZkController.stop()
Repository: samza
Updated Branches:
refs/heads/0.13.0 9f60e980b -> 94b2f9eec
SAMZA-1286: Close zk connection in ZkController.stop()
Author: Boris Shkolnik <bo...@apache.org>
Reviewers: Xinyu Liu <xi...@apache.org>
Closes #185 from sborya/ZkControllerClose
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/345cef89
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/345cef89
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/345cef89
Branch: refs/heads/0.13.0
Commit: 345cef89d30d73a760c66abce5d8456a71605e68
Parents: 9f60e98
Author: Boris Shkolnik <bo...@apache.org>
Authored: Thu May 11 11:35:55 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Mon May 15 14:37:06 2017 -0700
----------------------------------------------------------------------
.../src/main/java/org/apache/samza/zk/ZkControllerImpl.java | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/345cef89/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java b/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
index 0b831f8..44d2e37 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
@@ -76,6 +76,11 @@ public class ZkControllerImpl implements ZkController {
if (isLeader()) {
zkLeaderElector.resignLeadership();
}
+
+ // close zk connection
+ if (zkUtils != null) {
+ zkUtils.getZkClient().close();
+ }
}
@Override
[4/5] samza git commit: SAMZA-1289: Default id generator if not
configured
Posted by xi...@apache.org.
SAMZA-1289: Default id generator if not configured
Right now in standalone deployment we require the user to provide an id generator. Since most of the time the users can simply use the UUID generator for id generation, we should default it using that.
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Reviewers: Jagadish V <vj...@apache.org>
Closes #189 from xinyuiscool/SAMZA-1289
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/a0b5c53b
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/a0b5c53b
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/a0b5c53b
Branch: refs/heads/0.13.0
Commit: a0b5c53b73e434c58822e9ec80ef541fd30b1d2b
Parents: d562f98
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Authored: Fri May 12 14:20:15 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Mon May 15 14:37:45 2017 -0700
----------------------------------------------------------------------
.../src/main/java/org/apache/samza/operators/MessageStream.java | 1 +
.../main/java/org/apache/samza/config/ApplicationConfig.java | 5 ++++-
.../org/apache/samza/container/ContainerHeartbeatClient.java | 4 ++--
3 files changed, 7 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/a0b5c53b/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 b081869..bef1d3f 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
@@ -153,6 +153,7 @@ public interface MessageStream<M> {
* The merged {@link MessageStream} contains messages from all {@code streams} in the order they arrive.
*
* @param streams {@link MessageStream}s to be merged
+ * @param <T> the type of messages in each of the streams
* @return the merged {@link MessageStream}
* @throws IllegalArgumentException if {@code streams} is empty
*/
http://git-wip-us.apache.org/repos/asf/samza/blob/a0b5c53b/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java b/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
index 1434231..1b53321 100644
--- a/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
@@ -18,6 +18,9 @@
*/
package org.apache.samza.config;
+import org.apache.samza.runtime.UUIDGenerator;
+
+
/**
* Accessors for configs associated with Application scope
*/
@@ -52,7 +55,7 @@ public class ApplicationConfig extends MapConfig {
}
public String getAppProcessorIdGeneratorClass() {
- return get(APP_PROCESSOR_ID_GENERATOR_CLASS, null);
+ return get(APP_PROCESSOR_ID_GENERATOR_CLASS, UUIDGenerator.class.getName());
}
public String getAppName() {
http://git-wip-us.apache.org/repos/asf/samza/blob/a0b5c53b/samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatClient.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatClient.java b/samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatClient.java
index cc14948..f2c2651 100644
--- a/samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatClient.java
+++ b/samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatClient.java
@@ -60,8 +60,8 @@ public class ContainerHeartbeatClient {
}
/**
- * Issues a heartbeat request to the coordinator and
- * returns the corresponding {@link ContainerHeartbeatResponse}.
+ * Issues a heartbeat request to the coordinator
+ * @return the corresponding {@link ContainerHeartbeatResponse}.
*/
public ContainerHeartbeatResponse requestHeartbeat() {
ObjectMapper mapper = new ObjectMapper();
[5/5] samza git commit: SAMZA-1290: Fix javadoc for ApplicationRuner
Posted by xi...@apache.org.
SAMZA-1290: Fix javadoc for ApplicationRuner
Author: Jacob Maes <jm...@linkedin.com>
Reviewers: Prateek Maheshwari <pm...@linkedin.com>
Closes #190 from jmakes/samza-1290
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/94b2f9ee
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/94b2f9ee
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/94b2f9ee
Branch: refs/heads/0.13.0
Commit: 94b2f9eec4f21ab7d94c126124fe3c6fbf82f6bd
Parents: a0b5c53
Author: Jacob Maes <jm...@linkedin.com>
Authored: Fri May 12 15:22:47 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Mon May 15 14:38:01 2017 -0700
----------------------------------------------------------------------
.../src/main/java/org/apache/samza/runtime/ApplicationRunner.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/94b2f9ee/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java b/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
index eda09a2..0586e9e 100644
--- a/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
+++ b/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
@@ -29,7 +29,7 @@ import java.lang.reflect.Constructor;
/**
- * A physical execution engine to deploy the config and jobs to run the {@link org.apache.samza.operators.StreamGraph}
+ * The primary means of managing execution of the {@link org.apache.samza.application.StreamApplication} at runtime.
*/
@InterfaceStability.Unstable
public abstract class ApplicationRunner {
[3/5] samza git commit: SAMZA-1288: Add null check for sink
OutputStream
Posted by xi...@apache.org.
SAMZA-1288: Add null check for sink OutputStream
The logic to generate json for Sink operator does not check whether the output stream is null. This causes null pointer exception.
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Reviewers: Jake Maes <jm...@apached.org>
Closes #188 from xinyuiscool/SAMZA-1288
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/d562f986
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/d562f986
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/d562f986
Branch: refs/heads/0.13.0
Commit: d562f986e2240c83566fe2380336c1568f0c8f9d
Parents: b3b2fb5
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Authored: Fri May 12 09:49:04 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Mon May 15 14:37:38 2017 -0700
----------------------------------------------------------------------
.../apache/samza/operators/util/OperatorJsonUtils.java | 10 +++++-----
.../apache/samza/execution/TestJobGraphJsonGenerator.java | 3 ++-
2 files changed, 7 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/d562f986/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
index b52fbc3..b971607 100644
--- 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
@@ -27,12 +27,9 @@ 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.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.samza.operators.stream.OutputStreamInternal;
public class OperatorJsonUtils {
- private static final Logger log = LoggerFactory.getLogger(OperatorJsonUtils.class);
-
private static final String OP_CODE = "opCode";
private static final String OP_ID = "opId";
private static final String SOURCE_LOCATION = "sourceLocation";
@@ -59,7 +56,10 @@ public class OperatorJsonUtils {
}
if (spec instanceof SinkOperatorSpec) {
- map.put(OUTPUT_STREAM_ID, ((SinkOperatorSpec) spec).getOutputStream().getStreamSpec().getId());
+ OutputStreamInternal outputStream = ((SinkOperatorSpec) spec).getOutputStream();
+ if (outputStream != null) {
+ map.put(OUTPUT_STREAM_ID, outputStream.getStreamSpec().getId());
+ }
}
if (spec instanceof PartialJoinOperatorSpec) {
http://git-wip-us.apache.org/repos/asf/samza/blob/d562f986/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 2681f9c..e53cd42 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
@@ -113,6 +113,7 @@ public class TestJobGraphJsonGenerator {
OutputStream<Object, Object, Object> outputStream2 = streamGraph.getOutputStream("output2", mockFn, mockFn);
m1.join(m2, mock(JoinFunction.class), Duration.ofHours(2)).sendTo(outputStream1);
+ m2.sink((message, collector, coordinator) -> { });
m3.join(m2, mock(JoinFunction.class), Duration.ofHours(1)).sendTo(outputStream2);
ExecutionPlanner planner = new ExecutionPlanner(config, streamManager);
@@ -124,7 +125,7 @@ public class TestJobGraphJsonGenerator {
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() == 12);
+ assertTrue(nodes.jobs.get(0).operatorGraph.operators.size() == 13);
assertTrue(nodes.sourceStreams.size() == 3);
assertTrue(nodes.sinkStreams.size() == 2);
assertTrue(nodes.intermediateStreams.size() == 2);
[2/5] samza git commit: Increase the plan graph size
Posted by xi...@apache.org.
Increase the plan graph size
Increase the canvas size to a standard 24 inch resolution and also the scaling factor.
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Reviewers: Jake Maes <jm...@apache.org>
Closes #186 from xinyuiscool/PLAN
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/b3b2fb52
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/b3b2fb52
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/b3b2fb52
Branch: refs/heads/0.13.0
Commit: b3b2fb5210c1f5a8f615576701f801c2969df3b4
Parents: 345cef8
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Authored: Thu May 11 13:39:48 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Mon May 15 14:37:27 2017 -0700
----------------------------------------------------------------------
samza-shell/src/main/visualizer/plan.html | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/b3b2fb52/samza-shell/src/main/visualizer/plan.html
----------------------------------------------------------------------
diff --git a/samza-shell/src/main/visualizer/plan.html b/samza-shell/src/main/visualizer/plan.html
index 9fe2e26..e4fca66 100644
--- a/samza-shell/src/main/visualizer/plan.html
+++ b/samza-shell/src/main/visualizer/plan.html
@@ -83,7 +83,7 @@
<p id="summary"/>
</section>
-<svg id="svg-canvas" width=1200 height=1000><g/></svg>
+<svg id="svg-canvas" width=1920 height=1200><g/></svg>
<script id="js">
var data = JSON.parse(plan);
@@ -112,7 +112,7 @@
// Run the renderer. This is what draws the final graph.
render(inner, g);
- var initialScale = 0.65;
+ var initialScale = 0.75;
zoom.scale(initialScale).event(svg);
svg.attr('height', g.graph().height * initialScale + 100);
</script>