You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by vr...@apache.org on 2015/10/25 20:37:44 UTC
[01/36] incubator-apex-core git commit: APEX-132 #resolve added
Apache copyright template for IntelliJ
Repository: incubator-apex-core
Updated Branches:
refs/heads/feature-module c773c0e17 -> 4ce1b365e
APEX-132 #resolve added Apache copyright template for IntelliJ
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/50862129
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/50862129
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/50862129
Branch: refs/heads/feature-module
Commit: 5086212928f341581ca94cf415ae6ebd42a4bc31
Parents: b986f70
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Mon Oct 5 18:09:08 2015 -0700
Committer: Vlad Rozov <v....@datatorrent.com>
Committed: Tue Oct 6 22:43:36 2015 -0700
----------------------------------------------------------------------
.idea/copyright/apache.xml | 6 ++++++
.idea/copyright/profiles_settings.xml | 9 +++++++++
2 files changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/50862129/.idea/copyright/apache.xml
----------------------------------------------------------------------
diff --git a/.idea/copyright/apache.xml b/.idea/copyright/apache.xml
new file mode 100644
index 0000000..12f481d
--- /dev/null
+++ b/.idea/copyright/apache.xml
@@ -0,0 +1,6 @@
+<component name="CopyrightManager">
+ <copyright>
+ <option name="myName" value="apache" />
+ <option name="notice" value="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." />
+ </copyright>
+</component>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/50862129/.idea/copyright/profiles_settings.xml
----------------------------------------------------------------------
diff --git a/.idea/copyright/profiles_settings.xml b/.idea/copyright/profiles_settings.xml
new file mode 100644
index 0000000..70dfce5
--- /dev/null
+++ b/.idea/copyright/profiles_settings.xml
@@ -0,0 +1,9 @@
+<component name="CopyrightManager">
+ <settings default="apache">
+ <LanguageOptions name="__TEMPLATE__">
+ <option name="addBlankAfter" value="false" />
+ <option name="separateBefore" value="true" />
+ <option name="lenBefore" value="2" />
+ </LanguageOptions>
+ </settings>
+</component>
\ No newline at end of file
[28/36] incubator-apex-core git commit: APEX-212 #resolve Adding a
check for physical operators list being empty in fillOperatorInfo
Posted by vr...@apache.org.
APEX-212 #resolve Adding a check for physical operators list being empty in fillOperatorInfo
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/83f95d82
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/83f95d82
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/83f95d82
Branch: refs/heads/feature-module
Commit: 83f95d82db4549ea48be4965257605c147998187
Parents: e3cecc2
Author: ishark <is...@datatorrent.com>
Authored: Wed Oct 21 16:49:00 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Wed Oct 21 18:19:35 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/StreamingContainerManager.java | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/83f95d82/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
index 3931fad..ca724db 100644
--- a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
+++ b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
@@ -2396,9 +2396,12 @@ public class StreamingContainerManager implements PlanContext
}
}
}
- loi.checkpointTimeMA = checkpointTimeAggregate.getAvg().longValue();
- loi.counters = latestLogicalCounters.get(operator.getName());
- loi.autoMetrics = latestLogicalMetrics.get(operator.getName());
+ if (physicalOperators.size() > 0) {
+ loi.checkpointTimeMA = checkpointTimeAggregate.getAvg().longValue();
+ loi.counters = latestLogicalCounters.get(operator.getName());
+ loi.autoMetrics = latestLogicalMetrics.get(operator.getName());
+ }
+
return loi;
}
[23/36] incubator-apex-core git commit: Fix EOL
Posted by vr...@apache.org.
Fix EOL
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/42dd9f84
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/42dd9f84
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/42dd9f84
Branch: refs/heads/feature-module
Commit: 42dd9f84412675916bc1f0c487f106551b453d79
Parents: 72eda3d
Author: MalharJenkins <je...@datatorrent.com>
Authored: Tue Oct 20 19:54:58 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Tue Oct 20 20:34:30 2015 -0700
----------------------------------------------------------------------
.../datatorrent/common/util/package-info.java | 38 +++---
.../com/datatorrent/stram/cli/package-info.java | 66 ++++-----
.../datatorrent/stram/engine/package-info.java | 94 ++++++-------
.../com/datatorrent/stram/package-info.java | 54 ++++----
.../datatorrent/stram/stream/package-info.java | 136 +++++++++----------
.../datatorrent/stram/util/package-info.java | 64 ++++-----
.../datatorrent/stram/webapp/package-info.java | 76 +++++------
7 files changed, 264 insertions(+), 264 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/common/src/main/java/com/datatorrent/common/util/package-info.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/com/datatorrent/common/util/package-info.java b/common/src/main/java/com/datatorrent/common/util/package-info.java
index 41134d0..5bfc38f 100644
--- a/common/src/main/java/com/datatorrent/common/util/package-info.java
+++ b/common/src/main/java/com/datatorrent/common/util/package-info.java
@@ -1,22 +1,22 @@
-/**
- * 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.
- */
+/**
+ * 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.
+ */
/**
* Shared utilities
*/
-package com.datatorrent.common.util;
+package com.datatorrent.common.util;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/engine/src/main/java/com/datatorrent/stram/cli/package-info.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/cli/package-info.java b/engine/src/main/java/com/datatorrent/stram/cli/package-info.java
index 116366b..05b45e2 100644
--- a/engine/src/main/java/com/datatorrent/stram/cli/package-info.java
+++ b/engine/src/main/java/com/datatorrent/stram/cli/package-info.java
@@ -1,33 +1,33 @@
-/**
- * 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.
- */
-/**
- *
- * <b>com.datatorrent.stram.cli</b> package deals with code for command line interface for the streaming platform<p>
- * <br>
- * The cli code wraps the webservices layer (<b>com.datatorrent.stream.webapp</b>) and thus accesses the dag
- * via one single point. All webservice calls for streaming data go through to the stram.<br>
- * <br>
- * The com.datatorrent.stram.cli package consists of<br>
- * <b>{@link com.datatorrent.stram.cli.DTCli}</b>: Provides command line interface for a streaming application on hadoop (yarn)<br>
- * <br>
- *
- */
-package com.datatorrent.stram.cli;
-
-
+/**
+ * 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.
+ */
+/**
+ *
+ * <b>com.datatorrent.stram.cli</b> package deals with code for command line interface for the streaming platform<p>
+ * <br>
+ * The cli code wraps the webservices layer (<b>com.datatorrent.stream.webapp</b>) and thus accesses the dag
+ * via one single point. All webservice calls for streaming data go through to the stram.<br>
+ * <br>
+ * The com.datatorrent.stram.cli package consists of<br>
+ * <b>{@link com.datatorrent.stram.cli.DTCli}</b>: Provides command line interface for a streaming application on hadoop (yarn)<br>
+ * <br>
+ *
+ */
+package com.datatorrent.stram.cli;
+
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/engine/src/main/java/com/datatorrent/stram/engine/package-info.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/engine/package-info.java b/engine/src/main/java/com/datatorrent/stram/engine/package-info.java
index e7c1d6e..fdad92f 100644
--- a/engine/src/main/java/com/datatorrent/stram/engine/package-info.java
+++ b/engine/src/main/java/com/datatorrent/stram/engine/package-info.java
@@ -1,47 +1,47 @@
-/**
- * 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.
- */
-/**
- * Execution layer for data operators, tuple processing, serialization, streams, window boundaries etc.<p>
- * <br>
- * <br>
- * The streaming platform would take care of the following for you<br>(TBD, add "How it will")<br>
- * - Standard partitioning (round robin, sticky key). You can add your own partitioning.<br>
- * - End of window statistics collection in terms of number of tuples, bandwidth, I/O etc<br>
- * - Ensuring the the emitted tuple reaches the downstream operators<br>
- * - Queueing tuples and retaining them till all future downstream operators have consumed it<br>
- * <br>
- * <br>
- * There are pre-defined library operators that you can use: see ...<br>
- * Examples are in the following packages<br>
- <ul>
- <li><a href="../../../datatorrent/lib/algo/package-summary.html">Algorithmic</a>: A set of algorithmic operators</li>
- <li><a href="../../../datatorrent/lib/io/package-summary.html">Input-Output</a>: A set of operators for Input-Output from Hadoop. Consists of adapters to various message buses</li>
- <li><a href="../../../datatorrent/lib/logs/package-summary.html">Log Collection</a>: A set of operators for log collection</li>
- <li><a href="../../../datatorrent/lib/math/package-summary.html">Arithmetic</a>: A set of arithmetic operators</li>
- <li><a href="../../../datatorrent/lib/stream/package-summary.html">Stream</a>: A set of operators for stream operations</li>
- <li><a href="../../../datatorrent/lib/testbench/package-summary.html">Testbench</a>: A set of operators for testing your dag, operators </li>
- <li><a href="../../../datatorrent/lib/util/package-summary.html">Utilities</a>: A set of utility classes</li>
- </ul>
- * <br>
- *
- */
-
-package com.datatorrent.stram.engine;
-
-
+/**
+ * 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.
+ */
+/**
+ * Execution layer for data operators, tuple processing, serialization, streams, window boundaries etc.<p>
+ * <br>
+ * <br>
+ * The streaming platform would take care of the following for you<br>(TBD, add "How it will")<br>
+ * - Standard partitioning (round robin, sticky key). You can add your own partitioning.<br>
+ * - End of window statistics collection in terms of number of tuples, bandwidth, I/O etc<br>
+ * - Ensuring the the emitted tuple reaches the downstream operators<br>
+ * - Queueing tuples and retaining them till all future downstream operators have consumed it<br>
+ * <br>
+ * <br>
+ * There are pre-defined library operators that you can use: see ...<br>
+ * Examples are in the following packages<br>
+ <ul>
+ <li><a href="../../../datatorrent/lib/algo/package-summary.html">Algorithmic</a>: A set of algorithmic operators</li>
+ <li><a href="../../../datatorrent/lib/io/package-summary.html">Input-Output</a>: A set of operators for Input-Output from Hadoop. Consists of adapters to various message buses</li>
+ <li><a href="../../../datatorrent/lib/logs/package-summary.html">Log Collection</a>: A set of operators for log collection</li>
+ <li><a href="../../../datatorrent/lib/math/package-summary.html">Arithmetic</a>: A set of arithmetic operators</li>
+ <li><a href="../../../datatorrent/lib/stream/package-summary.html">Stream</a>: A set of operators for stream operations</li>
+ <li><a href="../../../datatorrent/lib/testbench/package-summary.html">Testbench</a>: A set of operators for testing your dag, operators </li>
+ <li><a href="../../../datatorrent/lib/util/package-summary.html">Utilities</a>: A set of utility classes</li>
+ </ul>
+ * <br>
+ *
+ */
+
+package com.datatorrent.stram.engine;
+
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/engine/src/main/java/com/datatorrent/stram/package-info.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/package-info.java b/engine/src/main/java/com/datatorrent/stram/package-info.java
index 3e23940..eb337d6 100644
--- a/engine/src/main/java/com/datatorrent/stram/package-info.java
+++ b/engine/src/main/java/com/datatorrent/stram/package-info.java
@@ -1,27 +1,27 @@
-/**
- * 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.
- */
-/**
- * <b>com.datatorrent.stram</b> package contains all code for streaming application master<p>
- * <br>
- * The application master is also called <b>STRAM</b><br>
- * (TBD - To explain all functionality)<br>
- *
- */
-
-package com.datatorrent.stram;
+/**
+ * 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.
+ */
+/**
+ * <b>com.datatorrent.stram</b> package contains all code for streaming application master<p>
+ * <br>
+ * The application master is also called <b>STRAM</b><br>
+ * (TBD - To explain all functionality)<br>
+ *
+ */
+
+package com.datatorrent.stram;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/engine/src/main/java/com/datatorrent/stram/stream/package-info.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/stream/package-info.java b/engine/src/main/java/com/datatorrent/stram/stream/package-info.java
index e83d345..58fd326 100644
--- a/engine/src/main/java/com/datatorrent/stram/stream/package-info.java
+++ b/engine/src/main/java/com/datatorrent/stram/stream/package-info.java
@@ -1,68 +1,68 @@
-/**
- * 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.
- */
-/**
- *
- * <b>com.datatorrent.stream</b> package contains all code related to various implementations of Stream interface<p>
- * <br>
- * A stream is a logical unit of a dag that defines the connection between
- * a node and list of listener operators. Stream has the following properties in Malhar's streaming platform<br>
- * - One writer node<br>
- * - Any number of listener operators<br>
- * - Context as defined by the properties specified in the dag<br>
- * A stream definition in the dag is a logical definition. Multiple logical listerner operators means that the emitted tuple
- * would reach each of them. Partitioning is done when a single
- * logical listener node partitions into multilpe physical operators. This may happen due to initial user
- * specification, or dynamic run time constraint enforcement. In such a scenerio the logical stream gets partitioned
- * into physical streams. Each physical stream would retain the
- * characteristics of the logical node (one writer, multiple readers, and context).<br>
- * <br>
- * The streams included in com.datatorrent.stream include<br>
- * <br>
- * <br><b>Buffer Server Streams</b><br>
- * <b>{@link com.datatorrent.stream.BufferServerInputStream}</b>: extends {@link com.datatorrent.stream.SocketInputStream},
- * takes data from buffer server into the node. Every logical stream will have at least two such
- * objects ({@link com.datatorrent.stream.BufferServerInputStream}
- * and {@link com.datatorrent.stream.BufferServerOutputStream}). If the logical stream gets partitioned
- * into physical streams then each of these physical streams will have these objects. Inlined version of
- * a logical stream does not go through the buffer server and hence would not have
- * {@link com.datatorrent.stream.BufferServerInputStream} and {@link com.datatorrent.stream.BufferServerOutputStream} objects<br>
- * <b>{@link com.datatorrent.stream.BufferServerOutputStream}</b>: extends {@link com.datatorrent.stream.SocketOutputStream}
- * and in conjunction with {@link com.datatorrent.stream.BufferServerInputStream} forms a complete stream
- * in a node->buffer server->node path<br>
- * <br><b>Inline Stream (Within a Hadoop Container)</b><br>
- * <b>{@link com.datatorrent.stream.InlineStream}</b>: Streams data between two operators in inline mode. This implementation of
- * {@link com.datatorrent.engine.Stream} and {{@link com.datatorrent.api.Sink}
- * interface does not have connection to BufferServer and cannot be persisted.<br>
- *
- * <b>{@link com.datatorrent.stream.MuxStream}</b>: <br>
- * <b>{@link com.datatorrent.stream.PartitionAwareSink}</b>: <br>
- *
- * <br><b>Socket Interface Streams</b><br>
- * <b>{@link com.datatorrent.stream.SocketInputStream}</b>: Implements {@link com.datatorrent.engine.Stream} interface and provides
- * basic stream connection for a node to read from a socket. Users can use this class if they want to directly connect to
- * a outside socket<br>
- * <b>{@link com.datatorrent.stream.SocketOutputStream}</b>: Implements {@link com.datatorrent.engine.Stream} interface and provides
- * basic stream connection for a node to write to a socket. Most likely users would not use it to write to a socket by themselves.
- * Would be used in adapters and via {@link com.datatorrent.stream.BufferServerOutputStream}<br>
- * <br>
- *
- */
-
-package com.datatorrent.stram.stream;
-
+/**
+ * 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.
+ */
+/**
+ *
+ * <b>com.datatorrent.stream</b> package contains all code related to various implementations of Stream interface<p>
+ * <br>
+ * A stream is a logical unit of a dag that defines the connection between
+ * a node and list of listener operators. Stream has the following properties in Malhar's streaming platform<br>
+ * - One writer node<br>
+ * - Any number of listener operators<br>
+ * - Context as defined by the properties specified in the dag<br>
+ * A stream definition in the dag is a logical definition. Multiple logical listerner operators means that the emitted tuple
+ * would reach each of them. Partitioning is done when a single
+ * logical listener node partitions into multilpe physical operators. This may happen due to initial user
+ * specification, or dynamic run time constraint enforcement. In such a scenerio the logical stream gets partitioned
+ * into physical streams. Each physical stream would retain the
+ * characteristics of the logical node (one writer, multiple readers, and context).<br>
+ * <br>
+ * The streams included in com.datatorrent.stream include<br>
+ * <br>
+ * <br><b>Buffer Server Streams</b><br>
+ * <b>{@link com.datatorrent.stream.BufferServerInputStream}</b>: extends {@link com.datatorrent.stream.SocketInputStream},
+ * takes data from buffer server into the node. Every logical stream will have at least two such
+ * objects ({@link com.datatorrent.stream.BufferServerInputStream}
+ * and {@link com.datatorrent.stream.BufferServerOutputStream}). If the logical stream gets partitioned
+ * into physical streams then each of these physical streams will have these objects. Inlined version of
+ * a logical stream does not go through the buffer server and hence would not have
+ * {@link com.datatorrent.stream.BufferServerInputStream} and {@link com.datatorrent.stream.BufferServerOutputStream} objects<br>
+ * <b>{@link com.datatorrent.stream.BufferServerOutputStream}</b>: extends {@link com.datatorrent.stream.SocketOutputStream}
+ * and in conjunction with {@link com.datatorrent.stream.BufferServerInputStream} forms a complete stream
+ * in a node->buffer server->node path<br>
+ * <br><b>Inline Stream (Within a Hadoop Container)</b><br>
+ * <b>{@link com.datatorrent.stream.InlineStream}</b>: Streams data between two operators in inline mode. This implementation of
+ * {@link com.datatorrent.engine.Stream} and {{@link com.datatorrent.api.Sink}
+ * interface does not have connection to BufferServer and cannot be persisted.<br>
+ *
+ * <b>{@link com.datatorrent.stream.MuxStream}</b>: <br>
+ * <b>{@link com.datatorrent.stream.PartitionAwareSink}</b>: <br>
+ *
+ * <br><b>Socket Interface Streams</b><br>
+ * <b>{@link com.datatorrent.stream.SocketInputStream}</b>: Implements {@link com.datatorrent.engine.Stream} interface and provides
+ * basic stream connection for a node to read from a socket. Users can use this class if they want to directly connect to
+ * a outside socket<br>
+ * <b>{@link com.datatorrent.stream.SocketOutputStream}</b>: Implements {@link com.datatorrent.engine.Stream} interface and provides
+ * basic stream connection for a node to write to a socket. Most likely users would not use it to write to a socket by themselves.
+ * Would be used in adapters and via {@link com.datatorrent.stream.BufferServerOutputStream}<br>
+ * <br>
+ *
+ */
+
+package com.datatorrent.stram.stream;
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/engine/src/main/java/com/datatorrent/stram/util/package-info.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/util/package-info.java b/engine/src/main/java/com/datatorrent/stram/util/package-info.java
index a5e1d50..200bf98 100644
--- a/engine/src/main/java/com/datatorrent/stram/util/package-info.java
+++ b/engine/src/main/java/com/datatorrent/stram/util/package-info.java
@@ -1,32 +1,32 @@
-/**
- * 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.
- */
-/**
- *
- * <b>com.datatorrent.util</b> package consists of utilities classes that Malhar provides<p>
- * <br>
- * The current utilities include<br>
- * <b>{@link com.datatorrent.util.CircularBuffer}</b>: Takes a class T and provides a circular buffer<br>
- * <b>{@link com.datatorrent.util.ScheduledExecutorService}</b>: <br>
- * <b>{@link com.datatorrent.util.ScheduledThreadPoolExecutor}</b>: <br>
- * <b>{@link com.datatorrent.util.StablePriorityQueue}</b>: Implements a priority queue (Queue<E>) and is mainly used to queue tuples<br>
- * <br>
- */
-
-package com.datatorrent.stram.util;
-
+/**
+ * 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.
+ */
+/**
+ *
+ * <b>com.datatorrent.util</b> package consists of utilities classes that Malhar provides<p>
+ * <br>
+ * The current utilities include<br>
+ * <b>{@link com.datatorrent.util.CircularBuffer}</b>: Takes a class T and provides a circular buffer<br>
+ * <b>{@link com.datatorrent.util.ScheduledExecutorService}</b>: <br>
+ * <b>{@link com.datatorrent.util.ScheduledThreadPoolExecutor}</b>: <br>
+ * <b>{@link com.datatorrent.util.StablePriorityQueue}</b>: Implements a priority queue (Queue<E>) and is mainly used to queue tuples<br>
+ * <br>
+ */
+
+package com.datatorrent.stram.util;
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/42dd9f84/engine/src/main/java/com/datatorrent/stram/webapp/package-info.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/package-info.java b/engine/src/main/java/com/datatorrent/stram/webapp/package-info.java
index a3d18bf..947f262 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/package-info.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/package-info.java
@@ -1,38 +1,38 @@
-/**
- * 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.
- */
-/**
- *
- * <b>com.datatorrent.stram.webapp</b> package is the web service layer of Malhar's streaming platform<p>
- * <br>
- * The webservices layer interacts with the streaming application master (stram). The internals of Hadoop are taken care
- * off and an common set of functionality is provided for all streaming related runtime data.<br>
- * <br>
- * The webservices layer consists of the following data:<br>
- * <b>{@link com.datatorrent.stram.webapp.AppInfo}</b>: Provides application level data like user, appId, elapsed time, etc.<br>
- * <b>{@link com.datatorrent.stram.webapp.OperatorInfo}</b>: Provides data on the operator. This includes throughput, container id etc.<br>
- * <b>{@link com.datatorrent.stram.webapp.OperatorsInfo}</b>: Provides data on all the operators of the data.<br>
- * <b>{@link com.datatorrent.stram.webapp.StramWebApp}</b>: TBD<br>
- * <b>{@link com.datatorrent.stram.webapp.StramWebServices}</b>: TBD<br>
- * <b>Access and Authoriation</b>: TBD<br>
- * <br>
- *
- */
-
-package com.datatorrent.stram.webapp;
-
+/**
+ * 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.
+ */
+/**
+ *
+ * <b>com.datatorrent.stram.webapp</b> package is the web service layer of Malhar's streaming platform<p>
+ * <br>
+ * The webservices layer interacts with the streaming application master (stram). The internals of Hadoop are taken care
+ * off and an common set of functionality is provided for all streaming related runtime data.<br>
+ * <br>
+ * The webservices layer consists of the following data:<br>
+ * <b>{@link com.datatorrent.stram.webapp.AppInfo}</b>: Provides application level data like user, appId, elapsed time, etc.<br>
+ * <b>{@link com.datatorrent.stram.webapp.OperatorInfo}</b>: Provides data on the operator. This includes throughput, container id etc.<br>
+ * <b>{@link com.datatorrent.stram.webapp.OperatorsInfo}</b>: Provides data on all the operators of the data.<br>
+ * <b>{@link com.datatorrent.stram.webapp.StramWebApp}</b>: TBD<br>
+ * <b>{@link com.datatorrent.stram.webapp.StramWebServices}</b>: TBD<br>
+ * <b>Access and Authoriation</b>: TBD<br>
+ * <br>
+ *
+ */
+
+package com.datatorrent.stram.webapp;
+
[31/36] incubator-apex-core git commit: APEX-218 Modify project names
for NOTICE.
Posted by vr...@apache.org.
APEX-218 Modify project names for NOTICE.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/62513101
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/62513101
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/62513101
Branch: refs/heads/feature-module
Commit: 625131017d4ba01cc542ff6d787f52c450c95abe
Parents: 3b3d068
Author: Thomas Weise <th...@datatorrent.com>
Authored: Thu Oct 22 20:35:32 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 23 12:45:05 2015 -0700
----------------------------------------------------------------------
apex-app-archetype/pom.xml | 2 +-
apex-conf-archetype/pom.xml | 2 +-
api/pom.xml | 2 +-
bufferserver/pom.xml | 2 +-
common/pom.xml | 2 +-
engine/pom.xml | 2 +-
pom.xml | 4 ++--
7 files changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/apex-app-archetype/pom.xml
----------------------------------------------------------------------
diff --git a/apex-app-archetype/pom.xml b/apex-app-archetype/pom.xml
index d9095cd..b70cfeb 100644
--- a/apex-app-archetype/pom.xml
+++ b/apex-app-archetype/pom.xml
@@ -31,7 +31,7 @@
<artifactId>apex-app-archetype</artifactId>
<packaging>maven-archetype</packaging>
- <name>Apache Apex Application Maven Archetype</name>
+ <name>Apache Apex (incubating) Application Maven Archetype</name>
<build>
<extensions>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/apex-conf-archetype/pom.xml
----------------------------------------------------------------------
diff --git a/apex-conf-archetype/pom.xml b/apex-conf-archetype/pom.xml
index 4af8d16..2e6f78b 100644
--- a/apex-conf-archetype/pom.xml
+++ b/apex-conf-archetype/pom.xml
@@ -31,7 +31,7 @@
<artifactId>apex-conf-archetype</artifactId>
<packaging>maven-archetype</packaging>
- <name>Apache Apex App Configuration Maven Archetype</name>
+ <name>Apache Apex (incubating) App Configuration Maven Archetype</name>
<build>
<extensions>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/api/pom.xml
----------------------------------------------------------------------
diff --git a/api/pom.xml b/api/pom.xml
index 9224e63..9447c78 100644
--- a/api/pom.xml
+++ b/api/pom.xml
@@ -30,7 +30,7 @@
</parent>
<artifactId>apex-api</artifactId>
- <name>Apache Apex API</name>
+ <name>Apache Apex (incubating) API</name>
<packaging>jar</packaging>
<repositories>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/bufferserver/pom.xml
----------------------------------------------------------------------
diff --git a/bufferserver/pom.xml b/bufferserver/pom.xml
index a45723b..acb93c9 100644
--- a/bufferserver/pom.xml
+++ b/bufferserver/pom.xml
@@ -31,7 +31,7 @@
<artifactId>apex-bufferserver</artifactId>
<description>Interface for nodes from 2 different containers to talk to each other.</description>
- <name>Apache Apex Buffer Server</name>
+ <name>Apache Apex (incubating) Buffer Server</name>
<build>
<plugins>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/common/pom.xml
----------------------------------------------------------------------
diff --git a/common/pom.xml b/common/pom.xml
index 1f7080d..6fb4852 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -29,7 +29,7 @@
</parent>
<artifactId>apex-common</artifactId>
- <name>Apache Apex Common Library</name>
+ <name>Apache Apex (incubating) Common Library</name>
<packaging>jar</packaging>
<build>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/engine/pom.xml
----------------------------------------------------------------------
diff --git a/engine/pom.xml b/engine/pom.xml
index 8e75151..5545e27 100644
--- a/engine/pom.xml
+++ b/engine/pom.xml
@@ -31,7 +31,7 @@
<artifactId>apex-engine</artifactId>
<packaging>jar</packaging>
- <name>Apache Apex Stream Processing Engine</name>
+ <name>Apache Apex (incubating) Stream Processing Engine</name>
<build>
<finalName>${project.artifactId}</finalName>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/62513101/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index f797805..05d66e0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -33,8 +33,8 @@
<version>3.3.0-incubating-SNAPSHOT</version>
<packaging>pom</packaging>
- <name>Apache Apex</name>
- <url>http://apex.apache.org</url>
+ <name>Apache Apex (incubating)</name>
+ <url>http://apex.incubator.apache.org</url>
<organization>
<name>Apache Software Foundation</name>
[11/36] incubator-apex-core git commit: APEX-199 #resolve Added
platform version check when launching app package
Posted by vr...@apache.org.
APEX-199 #resolve Added platform version check when launching app package
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/3e361866
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/3e361866
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/3e361866
Branch: refs/heads/feature-module
Commit: 3e3618668d672252a7b884fd80fb9da8c90378b9
Parents: 722fd67
Author: David Yan <da...@datatorrent.com>
Authored: Thu Oct 15 18:50:27 2015 -0700
Committer: David Yan <da...@datatorrent.com>
Committed: Thu Oct 15 18:50:27 2015 -0700
----------------------------------------------------------------------
.../java/com/datatorrent/stram/cli/DTCli.java | 18 ++++++++++++++++--
.../com/datatorrent/stram/util/VersionInfo.java | 19 +++++++++++++++++++
.../datatorrent/stram/util/VersionInfoTest.java | 12 ++++++++++++
3 files changed, 47 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3e361866/engine/src/main/java/com/datatorrent/stram/cli/DTCli.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/cli/DTCli.java b/engine/src/main/java/com/datatorrent/stram/cli/DTCli.java
index e126f73..89f92f0 100644
--- a/engine/src/main/java/com/datatorrent/stram/cli/DTCli.java
+++ b/engine/src/main/java/com/datatorrent/stram/cli/DTCli.java
@@ -1888,7 +1888,10 @@ public class DTCli
if (ap != null) {
try {
- checkCompatible(ap, cp);
+ if (!commandLineInfo.force) {
+ checkPlatformCompatible(ap);
+ checkConfigPackageCompatible(ap, cp);
+ }
launchAppPackage(ap, cp, commandLineInfo, reader);
return;
} finally {
@@ -3426,7 +3429,7 @@ public class DTCli
}
- private void checkCompatible(AppPackage ap, ConfigPackage cp)
+ private void checkConfigPackageCompatible(AppPackage ap, ConfigPackage cp)
{
if (cp == null) {
return;
@@ -3445,6 +3448,14 @@ public class DTCli
}
}
+ private void checkPlatformCompatible(AppPackage ap)
+ {
+ String apVersion = ap.getDtEngineVersion();
+ if (!VersionInfo.isCompatible(apVersion, VersionInfo.getVersion())) {
+ throw new CliException("This App Package is compiled with Apache Apex Core API version " + apVersion + ", which is incompatible with this Apex Core version " + VersionInfo.getVersion());
+ }
+ }
+
private void launchAppPackage(AppPackage ap, ConfigPackage cp, LaunchCommandLineInfo commandLineInfo, ConsoleReader reader) throws Exception
{
new LaunchCommand().execute(getLaunchAppPackageArgs(ap, cp, commandLineInfo, reader), reader);
@@ -3848,6 +3859,7 @@ public class DTCli
final Option originalAppID = add(OptionBuilder.withArgName("application id").hasArg().withDescription("Specify original application identifier for restart.").create("originalAppId"));
final Option exactMatch = add(new Option("exactMatch", "Only consider applications with exact app name"));
final Option queue = add(OptionBuilder.withArgName("queue name").hasArg().withDescription("Specify the queue to launch the application").create("queue"));
+ final Option force = add(new Option("force", "Force launch the application. Do not check for compatibility"));
private Option add(Option opt)
{
@@ -3888,6 +3900,7 @@ public class DTCli
result.args = line.getArgs();
result.origAppId = line.getOptionValue(LAUNCH_OPTIONS.originalAppID.getOpt());
result.exactMatch = line.hasOption("exactMatch");
+ result.force = line.hasOption("force");
return result;
}
@@ -3904,6 +3917,7 @@ public class DTCli
String archives;
String origAppId;
boolean exactMatch;
+ boolean force;
String[] args;
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3e361866/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java b/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
index cd1bc3b..fc29b01 100644
--- a/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
+++ b/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
@@ -155,6 +155,25 @@ public class VersionInfo {
}
}
+ public static boolean isCompatible(String thisVersion, String requiredVersion)
+ {
+ String[] thisVersionComponent = normalizeVersion(thisVersion).split("\\.");
+ String[] requiredVersionComponent = normalizeVersion(requiredVersion).split("\\.");
+
+ // major version check
+ if (!thisVersionComponent[0].equals(requiredVersionComponent[0])) {
+ return false;
+ }
+
+ // minor version check
+ if (Integer.parseInt(thisVersionComponent[1]) < Integer.parseInt(requiredVersionComponent[1])) {
+ return false;
+ }
+
+ // patch version doesn't matter
+ return true;
+ }
+
private static String normalizeVersion(String ver)
{
for (int i = 0; i < ver.length(); i++) {
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3e361866/engine/src/test/java/com/datatorrent/stram/util/VersionInfoTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/util/VersionInfoTest.java b/engine/src/test/java/com/datatorrent/stram/util/VersionInfoTest.java
index a7a2a1a..60d11a9 100644
--- a/engine/src/test/java/com/datatorrent/stram/util/VersionInfoTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/util/VersionInfoTest.java
@@ -42,4 +42,16 @@ public class VersionInfoTest
Assert.assertTrue(c < 0);
}
+ @Test
+ public void testCompatibleVersion()
+ {
+ Assert.assertFalse(VersionInfo.isCompatible("1.0", "1.1"));
+ Assert.assertTrue(VersionInfo.isCompatible("1.10", "1.2"));
+ Assert.assertTrue(VersionInfo.isCompatible("1.10.0", "1.10.34"));
+ Assert.assertTrue(VersionInfo.isCompatible("1.10.55", "1.10.3"));
+ Assert.assertTrue(VersionInfo.isCompatible("1.10.55", "1.10.55"));
+ Assert.assertFalse(VersionInfo.isCompatible("1.10.55", "2.10.55"));
+ Assert.assertFalse(VersionInfo.isCompatible("2.10.55", "1.10.55"));
+ }
+
}
[29/36] incubator-apex-core git commit: Merge branch 'devel-3' of
https://git-wip-us.apache.org/repos/asf/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'devel-3' of https://git-wip-us.apache.org/repos/asf/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/2e8ffc6b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/2e8ffc6b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/2e8ffc6b
Branch: refs/heads/feature-module
Commit: 2e8ffc6be9905dc2165965445a2122caa9b49776
Parents: eb0a95f 83f95d8
Author: Ilya Ganelin <il...@capitalone.com>
Authored: Wed Oct 21 22:07:47 2015 -0700
Committer: Ilya Ganelin <il...@capitalone.com>
Committed: Wed Oct 21 22:07:47 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/StreamingContainerManager.java | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
[36/36] incubator-apex-core git commit: Merge branch 'devel-3' of
https://github.com/apache/incubator-apex-core into feature-module
Posted by vr...@apache.org.
Merge branch 'devel-3' of https://github.com/apache/incubator-apex-core into feature-module
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/4ce1b365
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/4ce1b365
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/4ce1b365
Branch: refs/heads/feature-module
Commit: 4ce1b365e246548007122a2ad493ee663aba11f5
Parents: c773c0e 7f1f81f
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Sun Oct 25 12:36:34 2015 -0700
Committer: Vlad Rozov <v....@datatorrent.com>
Committed: Sun Oct 25 12:36:34 2015 -0700
----------------------------------------------------------------------
.idea/codeStyleSettings.xml | 102 +++++++
.idea/copyright/apache.xml | 6 +
.idea/copyright/profiles_settings.xml | 9 +
NOTICE | 6 +-
README.md | 15 +-
apex-app-archetype/README.md | 2 +-
apex-app-archetype/pom.xml | 8 +-
.../main/appended-resources/META-INF/NOTICE.vm | 3 +
.../META-INF/maven/archetype-metadata.xml | 6 -
.../XmlJavadocCommentsExtractor.xsl | 2 -
.../main/resources/archetype-resources/pom.xml | 25 +-
apex-conf-archetype/README.md | 2 +-
apex-conf-archetype/pom.xml | 8 +-
.../main/appended-resources/META-INF/NOTICE.vm | 3 +
apex_checks.xml | 14 +-
api/pom.xml | 14 +-
.../main/appended-resources/META-INF/NOTICE.vm | 3 +
bufferserver/pom.xml | 16 +-
.../main/appended-resources/META-INF/NOTICE.vm | 3 +
.../bufferserver/internal/DataList.java | 248 ++++++++++------
.../bufferserver/internal/DataListener.java | 2 +-
.../bufferserver/internal/FastDataList.java | 11 +-
.../bufferserver/internal/LogicalNode.java | 37 ++-
.../datatorrent/bufferserver/server/Server.java | 108 +++----
common/pom.xml | 14 +-
.../main/appended-resources/META-INF/NOTICE.vm | 3 +
.../common/util/AsyncFSStorageAgent.java | 10 +-
.../datatorrent/common/util/package-info.java | 38 +--
engine/pom.xml | 14 +-
.../main/appended-resources/META-INF/NOTICE.vm | 3 +
.../stram/StreamingContainerManager.java | 44 +--
.../java/com/datatorrent/stram/cli/DTCli.java | 18 +-
.../com/datatorrent/stram/cli/package-info.java | 66 ++---
.../datatorrent/stram/engine/package-info.java | 94 +++---
.../com/datatorrent/stram/package-info.java | 54 ++--
.../datatorrent/stram/stream/package-info.java | 136 ++++-----
.../com/datatorrent/stram/util/VersionInfo.java | 21 +-
.../datatorrent/stram/util/package-info.java | 64 ++--
.../stram/webapp/asm/CompactUtil.java | 14 +-
.../datatorrent/stram/webapp/package-info.java | 76 ++---
engine/src/main/scripts/dtcli | 2 +-
.../com/datatorrent/stram/CheckpointTest.java | 55 ++--
.../stram/LogicalPlanModificationTest.java | 34 +--
.../datatorrent/stram/OutputUnifiedTest.java | 29 +-
.../stram/StramLocalClusterTest.java | 17 +-
.../datatorrent/stram/StramMiniClusterTest.java | 6 +-
.../datatorrent/stram/StramRecoveryTest.java | 48 +--
.../com/datatorrent/stram/StreamCodecTest.java | 51 +---
.../stram/StreamingContainerManagerTest.java | 101 +++----
.../com/datatorrent/stram/cli/DTCliTest.java | 7 +-
.../stram/client/AppPackageTest.java | 2 +-
.../stram/engine/AutoMetricTest.java | 22 +-
.../stram/engine/ProcessingModeTests.java | 15 +-
.../stram/plan/StreamPersistanceTests.java | 27 +-
.../stram/support/StramTestSupport.java | 64 +++-
.../datatorrent/stram/util/VersionInfoTest.java | 12 +
.../resources/testAppPackage/.dt/dt-site.xml | 39 ---
.../test/resources/testAppPackage/dt-site.xml | 39 +++
.../resources/testAppPackage/mydtapp/pom.xml | 6 +-
header.txt | 16 +
license.txt | 16 -
misc/ide-templates/eclipse/README.md | 17 ++
.../eclipse/apex-importorder.importorder | 10 +
misc/ide-templates/eclipse/apex-style.xml | 295 +++++++++++++++++++
misc/ide-templates/intellij/README.md | 24 ++
misc/ide-templates/intellij/apex-style.jar | Bin 0 -> 1515 bytes
misc/ide-templates/netbeans/README.md | 14 +
misc/ide-templates/netbeans/apex-style.zip | Bin 0 -> 2278 bytes
pom.xml | 106 +++----
69 files changed, 1465 insertions(+), 931 deletions(-)
----------------------------------------------------------------------
[21/36] incubator-apex-core git commit: Merge remote-tracking branch
'david/APEX-199' into devel-3
Posted by vr...@apache.org.
Merge remote-tracking branch 'david/APEX-199' into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/72eda3d4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/72eda3d4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/72eda3d4
Branch: refs/heads/feature-module
Commit: 72eda3d4ca3543750e3ad2348071582647af4570
Parents: 2c5ccb0 3e36186
Author: Munagala V. Ramanath <ra...@apache.org>
Authored: Tue Oct 20 16:16:20 2015 -0700
Committer: Munagala V. Ramanath <ra...@apache.org>
Committed: Tue Oct 20 16:16:20 2015 -0700
----------------------------------------------------------------------
.../java/com/datatorrent/stram/cli/DTCli.java | 18 ++++++++++++++++--
.../com/datatorrent/stram/util/VersionInfo.java | 19 +++++++++++++++++++
.../datatorrent/stram/util/VersionInfoTest.java | 12 ++++++++++++
3 files changed, 47 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
[34/36] incubator-apex-core git commit: APEX-217 #resolve added
NOTICE.vm that generates original copyright in NOTICE. The workaround
suggested in https://issues.apache.org/jira/browse/MASFRES-5.
Posted by vr...@apache.org.
APEX-217 #resolve added NOTICE.vm that generates original copyright in NOTICE. The workaround suggested in https://issues.apache.org/jira/browse/MASFRES-5.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/b44c6468
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/b44c6468
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/b44c6468
Branch: refs/heads/feature-module
Commit: b44c6468f1b30570898c257c922543d9fb417dd6
Parents: a190e86
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Thu Oct 22 21:17:51 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 23 12:46:22 2015 -0700
----------------------------------------------------------------------
NOTICE | 1 +
apex-app-archetype/src/main/appended-resources/META-INF/NOTICE.vm | 3 +++
.../src/main/appended-resources/META-INF/NOTICE.vm | 3 +++
api/src/main/appended-resources/META-INF/NOTICE.vm | 3 +++
bufferserver/src/main/appended-resources/META-INF/NOTICE.vm | 3 +++
common/src/main/appended-resources/META-INF/NOTICE.vm | 3 +++
engine/src/main/appended-resources/META-INF/NOTICE.vm | 3 +++
pom.xml | 3 +++
8 files changed, 22 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/NOTICE
----------------------------------------------------------------------
diff --git a/NOTICE b/NOTICE
index eef5e86..1dfedde 100644
--- a/NOTICE
+++ b/NOTICE
@@ -4,6 +4,7 @@ Copyright (c) 2015 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
+
The initial developer of the original code is
DataTorrent, Inc. (http://www.datatorrent.com)
Copyright (c) 2012 - 2015. All Rights Reserved.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/apex-app-archetype/src/main/appended-resources/META-INF/NOTICE.vm
----------------------------------------------------------------------
diff --git a/apex-app-archetype/src/main/appended-resources/META-INF/NOTICE.vm b/apex-app-archetype/src/main/appended-resources/META-INF/NOTICE.vm
new file mode 100644
index 0000000..7edf492
--- /dev/null
+++ b/apex-app-archetype/src/main/appended-resources/META-INF/NOTICE.vm
@@ -0,0 +1,3 @@
+#if($project.properties.postNoticeText)
+$project.properties.postNoticeText
+#end
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/apex-conf-archetype/src/main/appended-resources/META-INF/NOTICE.vm
----------------------------------------------------------------------
diff --git a/apex-conf-archetype/src/main/appended-resources/META-INF/NOTICE.vm b/apex-conf-archetype/src/main/appended-resources/META-INF/NOTICE.vm
new file mode 100644
index 0000000..7edf492
--- /dev/null
+++ b/apex-conf-archetype/src/main/appended-resources/META-INF/NOTICE.vm
@@ -0,0 +1,3 @@
+#if($project.properties.postNoticeText)
+$project.properties.postNoticeText
+#end
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/api/src/main/appended-resources/META-INF/NOTICE.vm
----------------------------------------------------------------------
diff --git a/api/src/main/appended-resources/META-INF/NOTICE.vm b/api/src/main/appended-resources/META-INF/NOTICE.vm
new file mode 100644
index 0000000..7edf492
--- /dev/null
+++ b/api/src/main/appended-resources/META-INF/NOTICE.vm
@@ -0,0 +1,3 @@
+#if($project.properties.postNoticeText)
+$project.properties.postNoticeText
+#end
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/bufferserver/src/main/appended-resources/META-INF/NOTICE.vm
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/appended-resources/META-INF/NOTICE.vm b/bufferserver/src/main/appended-resources/META-INF/NOTICE.vm
new file mode 100644
index 0000000..7edf492
--- /dev/null
+++ b/bufferserver/src/main/appended-resources/META-INF/NOTICE.vm
@@ -0,0 +1,3 @@
+#if($project.properties.postNoticeText)
+$project.properties.postNoticeText
+#end
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/common/src/main/appended-resources/META-INF/NOTICE.vm
----------------------------------------------------------------------
diff --git a/common/src/main/appended-resources/META-INF/NOTICE.vm b/common/src/main/appended-resources/META-INF/NOTICE.vm
new file mode 100644
index 0000000..7edf492
--- /dev/null
+++ b/common/src/main/appended-resources/META-INF/NOTICE.vm
@@ -0,0 +1,3 @@
+#if($project.properties.postNoticeText)
+$project.properties.postNoticeText
+#end
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/engine/src/main/appended-resources/META-INF/NOTICE.vm
----------------------------------------------------------------------
diff --git a/engine/src/main/appended-resources/META-INF/NOTICE.vm b/engine/src/main/appended-resources/META-INF/NOTICE.vm
new file mode 100644
index 0000000..7edf492
--- /dev/null
+++ b/engine/src/main/appended-resources/META-INF/NOTICE.vm
@@ -0,0 +1,3 @@
+#if($project.properties.postNoticeText)
+$project.properties.postNoticeText
+#end
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b44c6468/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 05d66e0..9501807 100644
--- a/pom.xml
+++ b/pom.xml
@@ -108,6 +108,7 @@
<exclude>DISCLAIMER</exclude>
<exclude>LICENSE</exclude>
<exclude>NOTICE</exclude>
+ <exclude>**/NOTICE.vm</exclude>
<exclude>**/*.md</exclude>
<exclude>**/*.txt</exclude>
<exclude>**/*.importorder</exclude>
@@ -192,6 +193,7 @@
<exclude>**/*.txt</exclude>
<exclude>**/*.importorder</exclude>
<exclude>misc/ide-templates/**</exclude>
+ <exclude>**/NOTICE.vm</exclude>
</excludes>
</configuration>
</plugin>
@@ -362,6 +364,7 @@
<sourceReleaseAssemblyDescriptor>source-release-zip-tar</sourceReleaseAssemblyDescriptor>
<package.prefix>/opt/datatorrent</package.prefix>
<package.groupname>dtorrent</package.groupname>
+ <postNoticeText>The initial developer of the original code is
DataTorrent, Inc. (http://www.datatorrent.com)
Copyright (c) 2012 - 2015. All Rights Reserved.</postNoticeText>
</properties>
<distributionManagement>
<repository>
[35/36] incubator-apex-core git commit: APEX-219 Skip creation of
DEPENDENCIES for the source archive.
Posted by vr...@apache.org.
APEX-219 Skip creation of DEPENDENCIES for the source archive.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/7f1f81fd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/7f1f81fd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/7f1f81fd
Branch: refs/heads/feature-module
Commit: 7f1f81fd5326cd4ac776c33cb18c16d5cf41d90f
Parents: 40fba97
Author: Thomas Weise <th...@datatorrent.com>
Authored: Thu Oct 22 23:02:31 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 23 12:46:22 2015 -0700
----------------------------------------------------------------------
pom.xml | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/7f1f81fd/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 9501807..cbb6a81 100644
--- a/pom.xml
+++ b/pom.xml
@@ -394,6 +394,15 @@
</plugins>
</build>
</profile>
-
+ <profile>
+ <activation>
+ <file>
+ <missing>${basedir}/src/main</missing>
+ </file>
+ </activation>
+ <properties>
+ <remoteresources.skip>true</remoteresources.skip>
+ </properties>
+ </profile>
</profiles>
</project>
[22/36] incubator-apex-core git commit: APEX-184 #resolve 1. In
autoFlushExecutor don't exit run() until there is at least one listener that
has more data to send. 2. Do not enable read in resumeReadIfSuspended when
not able to switch to a new buffer. 3.
Posted by vr...@apache.org.
APEX-184 #resolve
1. In autoFlushExecutor don't exit run() until there is at least one listener that has more data to send.
2. Do not enable read in resumeReadIfSuspended when not able to switch to a new buffer.
3. Fix possible race condition in Block acquire.
4. Fix for incorrect counting of in memory block permits.
5. Fix check style violations.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/5b6e4281
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/5b6e4281
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/5b6e4281
Branch: refs/heads/feature-module
Commit: 5b6e4281865dd64baf8b12f87fd6b4225dcd0216
Parents: 48cd2e8
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Fri Oct 9 17:35:30 2015 -0700
Committer: Vlad Rozov <v....@datatorrent.com>
Committed: Tue Oct 20 19:22:02 2015 -0700
----------------------------------------------------------------------
bufferserver/pom.xml | 2 +-
.../bufferserver/internal/DataList.java | 248 ++++++++++++-------
.../bufferserver/internal/DataListener.java | 2 +-
.../bufferserver/internal/FastDataList.java | 11 +-
.../bufferserver/internal/LogicalNode.java | 37 ++-
.../datatorrent/bufferserver/server/Server.java | 108 ++++----
6 files changed, 232 insertions(+), 176 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/5b6e4281/bufferserver/pom.xml
----------------------------------------------------------------------
diff --git a/bufferserver/pom.xml b/bufferserver/pom.xml
index 939da4f..126e715 100644
--- a/bufferserver/pom.xml
+++ b/bufferserver/pom.xml
@@ -51,7 +51,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>124</maxAllowedViolations>
+ <maxAllowedViolations>60</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/5b6e4281/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataList.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataList.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataList.java
index f5af2e5..1f6c273 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataList.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataList.java
@@ -19,8 +19,13 @@
package com.datatorrent.bufferserver.internal;
import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
@@ -40,7 +45,6 @@ import com.datatorrent.bufferserver.util.VarInt;
import com.datatorrent.netlet.AbstractClient;
import com.datatorrent.netlet.util.VarInt.MutableInt;
-import static com.google.common.collect.Lists.newArrayList;
import static com.google.common.collect.Maps.newHashMap;
import static com.google.common.collect.Sets.newHashSet;
@@ -66,9 +70,10 @@ public class DataList
protected int size;
protected int processingOffset;
protected long baseSeconds;
- private final List<AbstractClient> suspendedClients = newArrayList();
+ private final Set<AbstractClient> suspendedClients = newHashSet();
private final AtomicInteger numberOfInMemBlockPermits;
private MutableInt nextOffset = new MutableInt();
+ private Future<?> future;
public DataList(final String identifier, final int blockSize, final int numberOfCacheBlocks)
{
@@ -106,13 +111,20 @@ public class DataList
for (Block temp = first; temp != null; temp = temp.next) {
if (temp.starting_window >= longWindowId || temp.ending_window > longWindowId) {
if (temp != last) {
+ last.refCount.decrementAndGet();
last = temp;
do {
temp = temp.next;
temp.discard(false);
- if (temp.data != null) {
- temp.data = null;
- numberOfInMemBlockRewound++;
+ synchronized (temp) {
+ if (temp.refCount.get() != 0) {
+ logger.debug("Discarded block {} has positive reference count. Listeners: {}", temp, all_listeners);
+ throw new IllegalStateException("Discarded block " + temp + " has positive reference count!");
+ }
+ if (temp.data != null) {
+ temp.data = null;
+ numberOfInMemBlockRewound++;
+ }
}
} while (temp.next != null);
last.next = null;
@@ -148,8 +160,13 @@ public class DataList
Block temp = first;
while (temp != last) {
temp.discard(false);
- temp.data = null;
- temp = temp.next;
+ synchronized (temp) {
+ if (temp.refCount.get() != 0) {
+ throw new IllegalStateException("Discarded block " + temp + " not zero reference count!");
+ }
+ temp.data = null;
+ temp = temp.next;
+ }
}
}
first = last;
@@ -173,9 +190,15 @@ public class DataList
break;
}
temp.discard(false);
- if (temp.data != null) {
- temp.data = null;
- numberOfInMemBlockPurged++;
+ synchronized (temp) {
+ if (temp.refCount.get() != 0) {
+ logger.debug("Discarded block {} has positive reference count. Listeners: {}", temp, all_listeners);
+ throw new IllegalStateException("Discarded block " + temp + " has positive reference count!");
+ }
+ if (temp.data != null) {
+ temp.data = null;
+ numberOfInMemBlockPurged++;
+ }
}
}
}
@@ -202,21 +225,15 @@ public class DataList
do {
while (size == 0) {
size = VarInt.read(last.data, processingOffset, writeOffset, nextOffset);
- switch (nextOffset.integer) {
- case -5:
- throw new RuntimeException("problemo!");
-
- case -4:
- case -3:
- case -2:
- case -1:
- case 0:
- if (writeOffset == last.data.length) {
- nextOffset.integer = 0;
- processingOffset = 0;
- size = 0;
- }
- break flush;
+ if (nextOffset.integer > -5 && nextOffset.integer < 1) {
+ if (writeOffset == last.data.length) {
+ nextOffset.integer = 0;
+ processingOffset = 0;
+ size = 0;
+ }
+ break flush;
+ } else if (nextOffset.integer == -5) {
+ throw new RuntimeException("problemo!");
}
}
@@ -240,6 +257,9 @@ public class DataList
Tuple rwt = Tuple.getTuple(last.data, processingOffset, size);
baseSeconds = (long)rwt.getBaseSeconds() << 32;
break;
+
+ default:
+ break;
}
processingOffset += size;
size = 0;
@@ -255,17 +275,28 @@ public class DataList
last.writingOffset = writeOffset;
- autoFlushExecutor.submit(new Runnable()
- {
- @Override
- public void run()
+ notifyListeners();
+
+ }
+
+ public void notifyListeners()
+ {
+ if (future == null || future.isDone() || future.isCancelled()) {
+ future = autoFlushExecutor.submit(new Runnable()
{
- for (DataListener dl : all_listeners) {
- dl.addedData();
+ @Override
+ public void run()
+ {
+ boolean atLeastOneListenerHasDataToSend;
+ do {
+ atLeastOneListenerHasDataToSend = false;
+ for (DataListener dl : all_listeners) {
+ atLeastOneListenerHasDataToSend |= dl.addedData();
+ }
+ } while (atLeastOneListenerHasDataToSend);
}
- }
-
- });
+ });
+ }
}
public void setAutoFlushExecutor(final ExecutorService es)
@@ -381,7 +412,7 @@ public class DataList
public boolean suspendRead(final AbstractClient client)
{
synchronized (suspendedClients) {
- return client.suspendReadIfResumed() && suspendedClients.add(client);
+ return suspendedClients.add(client) && client.suspendReadIfResumed();
}
}
@@ -395,6 +426,8 @@ public class DataList
}
suspendedClients.clear();
}
+ } else {
+ logger.debug("Keeping clients: {} suspended, numberOfInMemBlockPermits={}, Listeners: {}", suspendedClients, numberOfInMemBlockPermits, all_listeners);
}
return resumedSuspendedClients;
}
@@ -409,7 +442,7 @@ public class DataList
return new byte[blockSize];
}
- public void addBuffer(byte[] array)
+ public synchronized void addBuffer(byte[] array)
{
final int numberOfInMemBlockPermits = this.numberOfInMemBlockPermits.decrementAndGet();
if (numberOfInMemBlockPermits < 0) {
@@ -468,8 +501,7 @@ public class DataList
oldestBlockIndex = index;
oldestReadOffset = entry.getValue().getReadOffset();
status.slowestConsumer = entry.getKey();
- }
- else if (index == oldestBlockIndex && entry.getValue().getReadOffset() < oldestReadOffset) {
+ } else if (index == oldestBlockIndex && entry.getValue().getReadOffset() < oldestReadOffset) {
oldestReadOffset = entry.getValue().getReadOffset();
status.slowestConsumer = entry.getKey();
}
@@ -481,8 +513,7 @@ public class DataList
status.numBytesAllocated += b.data.length;
if (oldestBlockIndex == i) {
status.numBytesWaiting += b.writingOffset - oldestReadOffset;
- }
- else if (oldestBlockIndex < i) {
+ } else if (oldestBlockIndex < i) {
status.numBytesWaiting += b.writingOffset - b.readingOffset;
}
b = b.next;
@@ -508,7 +539,7 @@ public class DataList
/**
* actual data - stored as length followed by actual data.
*/
- byte data[];
+ byte[] data;
/**
* readingOffset is the offset of the first valid byte in the array.
*/
@@ -536,8 +567,8 @@ public class DataList
/**
* how count of references to this block.
*/
- AtomicInteger refCount;
- Future future;
+ private final AtomicInteger refCount;
+ private Future<?> future;
public Block(String id, int size)
{
@@ -566,8 +597,7 @@ public class DataList
if (current.offset + current.length > writingOffset) {
current.length = 0;
}
- }
- else {
+ } else {
current.length = 0;
}
}
@@ -581,7 +611,7 @@ public class DataList
SerializedData sd = dli.next();
switch (sd.buffer[sd.dataOffset]) {
case MessageType.RESET_WINDOW_VALUE:
- ResetWindowTuple rwt = (ResetWindowTuple) Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
+ ResetWindowTuple rwt = (ResetWindowTuple)Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
bs = (long)rwt.getBaseSeconds() << 32;
if (bs > windowId) {
writingOffset = sd.offset;
@@ -590,12 +620,15 @@ public class DataList
break;
case MessageType.BEGIN_WINDOW_VALUE:
- BeginWindowTuple bwt = (BeginWindowTuple) Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
+ BeginWindowTuple bwt = (BeginWindowTuple)Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
if ((bs | bwt.getWindowId()) >= windowId) {
writingOffset = sd.offset;
break done;
}
break;
+
+ default:
+ break;
}
}
}
@@ -628,19 +661,19 @@ public class DataList
SerializedData sd = dli.next();
switch (sd.buffer[sd.dataOffset]) {
case MessageType.RESET_WINDOW_VALUE:
- ResetWindowTuple rwt = (ResetWindowTuple) Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
+ ResetWindowTuple rwt = (ResetWindowTuple)Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
bs = (long)rwt.getBaseSeconds() << 32;
lastReset = sd;
break;
case MessageType.BEGIN_WINDOW_VALUE:
- BeginWindowTuple bwt = (BeginWindowTuple) Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
+ BeginWindowTuple bwt = (BeginWindowTuple)Tuple.getTuple(sd.buffer, sd.dataOffset, sd.length - sd.dataOffset + sd.offset);
if ((bs | bwt.getWindowId()) > longWindowId) {
found = true;
if (lastReset != null) {
- /*
- * Restore the last Reset tuple if there was any and adjust the writingOffset to the beginning of the reset tuple.
- */
+ /*
+ * Restore the last Reset tuple if there was any and adjust the writingOffset to the beginning of the reset tuple.
+ */
if (sd.offset >= lastReset.length) {
sd.offset -= lastReset.length;
if (!(sd.buffer == lastReset.buffer && sd.offset == lastReset.offset)) {
@@ -655,6 +688,10 @@ public class DataList
break done;
}
+ break;
+
+ default:
+ break;
}
}
}
@@ -671,8 +708,7 @@ public class DataList
System.arraycopy(lastReset.buffer, lastReset.offset, this.data, this.readingOffset, lastReset.length);
this.starting_window = this.ending_window = bs;
//logger.debug("=20140220= reassign the windowids {}", this);
- }
- else {
+ } else {
this.readingOffset = this.writingOffset;
this.starting_window = this.ending_window = longWindowId;
//logger.debug("=20140220= avoid the windowids {}", this);
@@ -692,8 +728,7 @@ public class DataList
sd.offset = 0;
sd.dataOffset = VarInt.write(sd.length - i, sd.buffer, sd.offset, i);
sd.buffer[sd.dataOffset] = MessageType.NO_MESSAGE_VALUE;
- }
- else {
+ } else {
logger.warn("Unhandled condition while purging the data purge to offset {}", sd.offset);
}
@@ -710,15 +745,17 @@ public class DataList
{
byte[] data = storage.retrieve(identifier, uniqueIdentifier);
synchronized (Block.this) {
- Block.this.data = data;
- readingOffset = 0;
- writingOffset = data.length;
- if (refCount.get() > 1) {
+ if (Block.this.data == null) {
+ Block.this.data = data;
+ readingOffset = 0;
+ writingOffset = data.length;
Block.this.notifyAll();
- }
- int numberOfInMemBlockPermits = DataList.this.numberOfInMemBlockPermits.decrementAndGet();
- if (numberOfInMemBlockPermits < 0) {
- logger.warn("Exceeded allowed memory block allocation by {}", -numberOfInMemBlockPermits);
+ int numberOfInMemBlockPermits = DataList.this.numberOfInMemBlockPermits.decrementAndGet();
+ if (numberOfInMemBlockPermits < 0) {
+ logger.warn("Exceeded allowed memory block allocation by {}", -numberOfInMemBlockPermits);
+ }
+ } else {
+ logger.debug("Block {} was already loaded into memory", Block.this);
}
}
}
@@ -727,20 +764,34 @@ public class DataList
protected void acquire(boolean wait)
{
- if (refCount.getAndIncrement() == 0 && storage != null && data == null) {
+ int refCount = this.refCount.getAndIncrement();
+ synchronized (Block.this) {
+ if (data != null) {
+ return;
+ }
+ }
+ if (refCount == 0 && storage != null) {
final Runnable retriever = getRetriever();
+ if (future != null && future.cancel(false)) {
+ logger.debug("Block {} future is cancelled", this);
+ }
if (wait) {
+ future = null;
retriever.run();
} else {
future = storageExecutor.submit(retriever);
}
- } else if (wait && data == null) {
+ } else if (wait) {
try {
synchronized (Block.this) {
- wait();
+ if (future == null) {
+ throw new IllegalStateException("No task is scheduled to retrieve block " + Block.this);
+ }
+ while (data == null) {
+ wait();
+ }
}
- }
- catch (InterruptedException ex) {
+ } catch (InterruptedException ex) {
throw new RuntimeException("Interrupted while waiting for data to be loaded!", ex);
}
}
@@ -758,15 +809,16 @@ public class DataList
}
if (uniqueIdentifier == 0) {
logger.warn("Storage returned unexpectedly, please check the status of the spool directory!");
- }
- else {
- //logger.debug("Spooled {} to disk", Block.this);
+ } else {
+ int numberOfInMemBlockPermits = DataList.this.numberOfInMemBlockPermits.get();
synchronized (Block.this) {
- if (refCount.get() == 0) {
+ if (refCount.get() == 0 && Block.this.data != null) {
Block.this.data = null;
+ numberOfInMemBlockPermits = DataList.this.numberOfInMemBlockPermits.incrementAndGet();
+ } else {
+ logger.debug("Keeping Block {} unchanged", Block.this);
}
}
- int numberOfInMemBlockPermits = DataList.this.numberOfInMemBlockPermits.incrementAndGet();
assert numberOfInMemBlockPermits < MAX_COUNT_OF_INMEM_BLOCKS : "Number of in memory block permits " + numberOfInMemBlockPermits + " exceeded configured maximum " + MAX_COUNT_OF_INMEM_BLOCKS + '.';
resumeSuspendedClients(numberOfInMemBlockPermits);
}
@@ -780,10 +832,17 @@ public class DataList
if (refCount == 0 && storage != null) {
assert (next != null);
final Runnable storer = getStorer(data, readingOffset, writingOffset, storage);
- if (wait && numberOfInMemBlockPermits.get() == 0) {
+ if (future != null && future.cancel(false)) {
+ logger.debug("Block {} future is cancelled", this);
+ }
+ final int numberOfInMemBlockPermits = DataList.this.numberOfInMemBlockPermits.get();
+ if (wait && numberOfInMemBlockPermits == 0) {
+ future = null;
storer.run();
- } else if (numberOfInMemBlockPermits.get() < MAX_COUNT_OF_INMEM_BLOCKS/2) {
+ } else if (numberOfInMemBlockPermits < MAX_COUNT_OF_INMEM_BLOCKS / 2) {
future = storageExecutor.submit(storer);
+ } else {
+ future = null;
}
} else {
logger.debug("Holding {} in memory due to {} references.", this, refCount);
@@ -809,11 +868,12 @@ public class DataList
protected void discard(final boolean wait)
{
if (storage != null) {
- if (future != null) {
- future.cancel(false);
- }
final Runnable discarder = getDiscarder();
+ if (future != null && future.cancel(false)) {
+ logger.debug("Block {} future is cancelled", this);
+ }
if (wait) {
+ future = null;
discarder.run();
} else {
future = storageExecutor.submit(discarder);
@@ -828,7 +888,7 @@ public class DataList
+ ", readingOffset=" + readingOffset + ", writingOffset=" + writingOffset
+ ", starting_window=" + Codec.getStringWindowId(starting_window) + ", ending_window=" + Codec.getStringWindowId(ending_window)
+ ", refCount=" + refCount.get() + ", uniqueIdentifier=" + uniqueIdentifier + ", next=" + (next == null ? "null" : next.identifier)
- + '}';
+ + ", future=" + (future == null ? "null" : future.isDone() ? "Done" : future.isCancelled() ? "Cancelled" : future) + '}';
}
}
@@ -895,19 +955,13 @@ public class DataList
{
while (size == 0) {
size = VarInt.read(buffer, readOffset, da.writingOffset, nextOffset);
- switch (nextOffset.integer) {
- case -5:
- throw new RuntimeException("problemo!");
-
- case -4:
- case -3:
- case -2:
- case -1:
- case 0:
- if (da.writingOffset == buffer.length && switchToNextBlock()) {
- continue;
- }
- return false;
+ if (nextOffset.integer > -5 && nextOffset.integer < 1) {
+ if (da.writingOffset == buffer.length && switchToNextBlock()) {
+ continue;
+ }
+ return false;
+ } else if (size == -5) {
+ throw new RuntimeException("problemo!");
}
}
@@ -965,6 +1019,12 @@ public class DataList
size = 0;
}
+ @Override
+ public String toString()
+ {
+ return getClass().getName() + '@' + Integer.toHexString(hashCode()) + "{da=" + da + '}';
+ }
+
}
private static final Logger logger = LoggerFactory.getLogger(DataList.class);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/5b6e4281/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataListener.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataListener.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataListener.java
index fd9cebc..4add008 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataListener.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/DataListener.java
@@ -36,7 +36,7 @@ public interface DataListener
/**
*/
- public void addedData();
+ public boolean addedData();
/**
*
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/5b6e4281/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
index 939d0c1..6ba7b64 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
@@ -99,17 +99,8 @@ public class FastDataList extends DataList
last.writingOffset = writeOffset;
- autoFlushExecutor.submit(new Runnable()
- {
- @Override
- public void run()
- {
- for (DataListener dl : all_listeners) {
- dl.addedData();
- }
- }
+ notifyListeners();
- });
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/5b6e4281/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/LogicalNode.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/LogicalNode.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/LogicalNode.java
index 40a8207..f867d69 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/LogicalNode.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/LogicalNode.java
@@ -73,8 +73,7 @@ public class LogicalNode implements DataListener
if (iterator instanceof DataListIterator) {
this.iterator = (DataListIterator)iterator;
- }
- else {
+ } else {
throw new IllegalArgumentException("iterator does not belong to DataListIterator class");
}
@@ -195,12 +194,12 @@ public class LogicalNode implements DataListener
case MessageType.BEGIN_WINDOW_VALUE:
tuple = Tuple.getTuple(data.buffer, data.dataOffset, data.length - data.dataOffset + data.offset);
logger.debug("{}->{} condition {} =? {}",
- new Object[] {
- upstream,
- group,
- Codec.getStringWindowId(baseSeconds | tuple.getWindowId()),
- Codec.getStringWindowId(skipWindowId)
- });
+ new Object[] {
+ upstream,
+ group,
+ Codec.getStringWindowId(baseSeconds | tuple.getWindowId()),
+ Codec.getStringWindowId(skipWindowId)
+ });
if ((baseSeconds | tuple.getWindowId()) > skipWindowId) {
logger.debug("caught up {}->{} skipping {} payload tuples", upstream, group, skippedPayloadTuples);
ready = GiveAll.getInstance().distribute(physicalNodes, data);
@@ -219,8 +218,7 @@ public class LogicalNode implements DataListener
logger.debug("Message {} was not distributed to {}", MessageType.valueOf(data.buffer[data.dataOffset]), physicalNodes);
}
}
- }
- catch (InterruptedException ie) {
+ } catch (InterruptedException ie) {
throw new RuntimeException(ie);
}
@@ -232,9 +230,8 @@ public class LogicalNode implements DataListener
logger.debug("Exiting catch up because caughtup = {}", caughtup);
}
- @SuppressWarnings("fallthrough")
@Override
- public void addedData()
+ public boolean addedData()
{
if (isReady()) {
if (caughtup) {
@@ -257,6 +254,8 @@ public class LogicalNode implements DataListener
case MessageType.RESET_WINDOW_VALUE:
Tuple resetWindow = Tuple.getTuple(data.buffer, data.dataOffset, data.length - data.dataOffset + data.offset);
baseSeconds = (long)resetWindow.getBaseSeconds() << 32;
+ ready = GiveAll.getInstance().distribute(physicalNodes, data);
+ break;
default:
//logger.debug("sending data of type {}", MessageType.valueOf(data.buffer[data.dataOffset]));
@@ -264,8 +263,7 @@ public class LogicalNode implements DataListener
break;
}
}
- }
- else {
+ } else {
while (ready && iterator.hasNext()) {
SerializedData data = iterator.next();
switch (data.buffer[data.dataOffset]) {
@@ -287,6 +285,8 @@ public class LogicalNode implements DataListener
case MessageType.RESET_WINDOW_VALUE:
tuple = Tuple.getTuple(data.buffer, data.dataOffset, data.length - data.dataOffset + data.offset);
baseSeconds = (long)tuple.getBaseSeconds() << 32;
+ ready = GiveAll.getInstance().distribute(physicalNodes, data);
+ break;
default:
ready = GiveAll.getInstance().distribute(physicalNodes, data);
@@ -294,15 +294,14 @@ public class LogicalNode implements DataListener
}
}
}
- }
- catch (InterruptedException ie) {
+ } catch (InterruptedException ie) {
throw new RuntimeException(ie);
}
- }
- else {
+ } else {
catchUp();
}
}
+ return !ready;
}
/**
@@ -345,7 +344,7 @@ public class LogicalNode implements DataListener
@Override
public String toString()
{
- return "LogicalNode{" + "upstream=" + upstream + ", group=" + group + ", partitions=" + partitions + '}';
+ return "LogicalNode{" + "upstream=" + upstream + ", group=" + group + ", partitions=" + partitions + ", iterator=" + iterator + '}';
}
private static final Logger logger = LoggerFactory.getLogger(LogicalNode.class);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/5b6e4281/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
index 9f31e02..cd45738 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
@@ -29,7 +29,13 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
-import java.util.concurrent.*;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +43,12 @@ import org.slf4j.LoggerFactory;
import com.datatorrent.bufferserver.internal.DataList;
import com.datatorrent.bufferserver.internal.FastDataList;
import com.datatorrent.bufferserver.internal.LogicalNode;
-import com.datatorrent.bufferserver.packet.*;
+import com.datatorrent.bufferserver.packet.PayloadTuple;
+import com.datatorrent.bufferserver.packet.PublishRequestTuple;
+import com.datatorrent.bufferserver.packet.PurgeRequestTuple;
+import com.datatorrent.bufferserver.packet.ResetRequestTuple;
+import com.datatorrent.bufferserver.packet.SubscribeRequestTuple;
+import com.datatorrent.bufferserver.packet.Tuple;
import com.datatorrent.bufferserver.storage.Storage;
import com.datatorrent.common.util.NameableThreadFactory;
import com.datatorrent.netlet.AbstractLengthPrependerClient;
@@ -100,16 +111,15 @@ public class Server implements ServerListener
@Override
public void unregistered(SelectionKey key)
{
- serverHelperExecutor.shutdown();
- storageHelperExecutor.shutdown();
- try {
- serverHelperExecutor.awaitTermination(5000, TimeUnit.MILLISECONDS);
- }
- catch (InterruptedException ex) {
- logger.debug("Executor Termination", ex);
- }
- logger.info("Server stopped listening at {}", address);
- }
+ serverHelperExecutor.shutdown();
+ storageHelperExecutor.shutdown();
+ try {
+ serverHelperExecutor.awaitTermination(5000, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException ex) {
+ logger.debug("Executor Termination", ex);
+ }
+ logger.info("Server stopped listening at {}", address);
+ }
public synchronized InetSocketAddress run(EventLoop eventloop)
{
@@ -117,8 +127,7 @@ public class Server implements ServerListener
while (address == null) {
try {
wait(20);
- }
- catch (InterruptedException ex) {
+ } catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}
@@ -142,8 +151,7 @@ public class Server implements ServerListener
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
- }
- else {
+ } else {
port = 0;
}
@@ -173,8 +181,7 @@ public class Server implements ServerListener
byte[] message;
if (dl == null) {
message = ("Invalid identifier '" + request.getIdentifier() + "'").getBytes();
- }
- else {
+ } else {
dl.purge(request.getBaseSeconds(), request.getWindowId());
message = ("Request sent for processing: " + request).getBytes();
}
@@ -201,8 +208,7 @@ public class Server implements ServerListener
byte[] message;
if (dl == null) {
message = ("Invalid identifier '" + request.getIdentifier() + "'").getBytes();
- }
- else {
+ } else {
AbstractLengthPrependerClient channel = publisherChannels.remove(request.getIdentifier());
if (channel != null) {
eventloop.disconnect(channel);
@@ -252,8 +258,7 @@ public class Server implements ServerListener
ln = subscriberGroups.get(type);
ln.boot(eventloop);
ln.addConnection(connection);
- }
- else {
+ } else {
/*
* if there is already a datalist registered for the type in which this client is interested,
* then get a iterator on the data items of that data list. If the datalist is not registered,
@@ -263,8 +268,7 @@ public class Server implements ServerListener
if (publisherBuffers.containsKey(upstream_identifier)) {
dl = publisherBuffers.get(upstream_identifier);
//logger.debug("old list = {}", dl);
- }
- else {
+ } else {
dl = Tuple.FAST_VERSION.equals(request.getVersion()) ? new FastDataList(upstream_identifier, blockSize, numberOfCacheBlocks) : new DataList(upstream_identifier, blockSize, numberOfCacheBlocks);
publisherBuffers.put(upstream_identifier, dl);
//logger.debug("new list = {}", dl);
@@ -315,12 +319,10 @@ public class Server implements ServerListener
dl = publisherBuffers.get(identifier);
try {
dl.rewind(request.getBaseSeconds(), request.getWindowId());
- }
- catch (IOException ie) {
+ } catch (IOException ie) {
throw new RuntimeException(ie);
}
- }
- else {
+ } else {
dl = Tuple.FAST_VERSION.equals(request.getVersion()) ? new FastDataList(identifier, blockSize, numberOfCacheBlocks) : new DataList(identifier, blockSize, numberOfCacheBlocks);
publisherBuffers.put(identifier, dl);
}
@@ -422,8 +424,7 @@ public class Server implements ServerListener
}
};
- }
- else {
+ } else {
publisher = new Publisher(dl, (long)request.getBaseSeconds() << 32 | request.getWindowId());
}
@@ -457,8 +458,7 @@ public class Server implements ServerListener
// }
if (subscriberRequest.getVersion().equals(Tuple.FAST_VERSION)) {
subscriber = new Subscriber(subscriberRequest.getStreamType(), subscriberRequest.getMask(), subscriberRequest.getPartitions(), bufferSize);
- }
- else {
+ } else {
subscriber = new Subscriber(subscriberRequest.getStreamType(), subscriberRequest.getMask(), subscriberRequest.getPartitions(), bufferSize)
{
@Override
@@ -494,8 +494,7 @@ public class Server implements ServerListener
logger.info("Received purge request: {}", request);
try {
handlePurgeRequest((PurgeRequestTuple)request, this);
- }
- catch (IOException io) {
+ } catch (IOException io) {
throw new RuntimeException(io);
}
break;
@@ -504,8 +503,7 @@ public class Server implements ServerListener
logger.info("Received reset all request: {}", request);
try {
handleResetRequest((ResetRequestTuple)request, this);
- }
- catch (IOException io) {
+ } catch (IOException io) {
throw new RuntimeException(io);
}
break;
@@ -636,9 +634,13 @@ public class Server implements ServerListener
{
final int interestOps = key.interestOps();
if ((interestOps & SelectionKey.OP_READ) == 0) {
- logger.debug("Resuming read on key {} with attachment {}", key, key.attachment());
- read(0);
- key.interestOps(interestOps | SelectionKey.OP_READ);
+ if (readExt(0)) {
+ logger.debug("Resuming read on key {} with attachment {}", key, key.attachment());
+ key.interestOps(interestOps | SelectionKey.OP_READ);
+ } else {
+ logger.debug("Keeping read on key {} with attachment {} suspended. ", key, key.attachment(), datalist);
+ datalist.notifyListeners();
+ }
}
}
});
@@ -648,6 +650,11 @@ public class Server implements ServerListener
@Override
public void read(int len)
{
+ readExt(len);
+ }
+
+ private boolean readExt(int len)
+ {
//logger.debug("read {} bytes", len);
writeOffset += len;
do {
@@ -664,18 +671,20 @@ public class Server implements ServerListener
* new byteBuffer and start as if we always had full room but not enough data.
*/
if (!switchToNewBufferOrSuspendRead(buffer, readOffset)) {
- return;
+ return false;
}
}
- }
- else if (dirty) {
+ } else if (dirty) {
dirty = false;
datalist.flush(writeOffset);
}
- return;
+ return true;
case 0:
continue;
+
+ default:
+ break;
}
}
@@ -683,8 +692,7 @@ public class Server implements ServerListener
onMessage(buffer, readOffset, size);
readOffset += size;
size = 0;
- }
- else {
+ } else {
if (writeOffset == buffer.length) {
dirty = false;
datalist.flush(writeOffset);
@@ -694,14 +702,14 @@ public class Server implements ServerListener
if (!switchToNewBufferOrSuspendRead(buffer, readOffset - VarInt.getSize(size))) {
readOffset -= VarInt.getSize(size);
size = 0;
- return;
+ return false;
}
size = 0;
} else if (dirty) {
dirty = false;
datalist.flush(writeOffset);
}
- return;
+ return true;
}
}
while (true);
@@ -751,8 +759,7 @@ public class Server implements ServerListener
if (cce instanceof RejectedExecutionException && serverHelperExecutor.isTerminated()) {
logger.warn("Terminated Executor Exception for {}.", this, cce);
el.disconnect(this);
- }
- else {
+ } else {
super.handleException(cce, el);
}
}
@@ -836,8 +843,7 @@ public class Server implements ServerListener
if (len < remainingCapacity) {
remainingCapacity = len;
byteBuffer.position(writeOffset + remainingCapacity);
- }
- else {
+ } else {
byteBuffer.position(buffer.length);
}
System.arraycopy(array, offset, buffer, writeOffset, remainingCapacity);
[20/36] incubator-apex-core git commit: APEX-198 Modify test to
create user directory under target.
Posted by vr...@apache.org.
APEX-198 Modify test to create user directory under target.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/2c5ccb01
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/2c5ccb01
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/2c5ccb01
Branch: refs/heads/feature-module
Commit: 2c5ccb0166addee37cedaa495de29824e0f499a0
Parents: 15ff42d
Author: David Yan <da...@datatorrent.com>
Authored: Tue Oct 20 14:50:02 2015 -0700
Committer: David Yan <da...@datatorrent.com>
Committed: Tue Oct 20 15:37:04 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/cli/DTCliTest.java | 7 +++-
.../resources/testAppPackage/.dt/dt-site.xml | 39 --------------------
.../test/resources/testAppPackage/dt-site.xml | 39 ++++++++++++++++++++
3 files changed, 44 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/2c5ccb01/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java b/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
index f8e1789..59a45aa 100644
--- a/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
@@ -64,7 +64,10 @@ public class DTCliTest
{
try {
userHome = System.getProperty("user.home");
- env.put("HOME", System.getProperty("user.dir") + "/src/test/resources/testAppPackage");
+ String newHome = System.getProperty("user.dir") + "/target";
+ FileUtils.forceMkdir(new File(newHome + "/.dt"));
+ FileUtils.copyFile(new File(System.getProperty("user.dir") + "/src/test/resources/testAppPackage/dt-site.xml"), new File(newHome + "/.dt/dt-site.xml"));
+ env.put("HOME", newHome);
setEnv(env);
cli.init();
@@ -83,7 +86,7 @@ public class DTCliTest
public static void finished()
{
try {
- env.put("HOME", System.getProperty("user.dir") + userHome);
+ env.put("HOME", userHome);
setEnv(env);
StramTestSupport.removeAppPackageFile();
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/2c5ccb01/engine/src/test/resources/testAppPackage/.dt/dt-site.xml
----------------------------------------------------------------------
diff --git a/engine/src/test/resources/testAppPackage/.dt/dt-site.xml b/engine/src/test/resources/testAppPackage/.dt/dt-site.xml
deleted file mode 100644
index f247ac9..0000000
--- a/engine/src/test/resources/testAppPackage/.dt/dt-site.xml
+++ /dev/null
@@ -1,39 +0,0 @@
-<?xml version="1.0" encoding="UTF-8" standalone="no"?>
-<!--
-
- 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.
-
--->
-<configuration>
- <property>
- <name>dt.test.1</name>
- <value>user-home-config</value>
- </property>
- <property>
- <name>dt.test.2</name>
- <value>user-home-config</value>
- </property>
- <property>
- <name>dt.test.3</name>
- <value>user-home-config</value>
- </property>
- <property>
- <name>dt.test.4</name>
- <value>user-home-config</value>
- </property>
-</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/2c5ccb01/engine/src/test/resources/testAppPackage/dt-site.xml
----------------------------------------------------------------------
diff --git a/engine/src/test/resources/testAppPackage/dt-site.xml b/engine/src/test/resources/testAppPackage/dt-site.xml
new file mode 100644
index 0000000..f247ac9
--- /dev/null
+++ b/engine/src/test/resources/testAppPackage/dt-site.xml
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!--
+
+ 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.
+
+-->
+<configuration>
+ <property>
+ <name>dt.test.1</name>
+ <value>user-home-config</value>
+ </property>
+ <property>
+ <name>dt.test.2</name>
+ <value>user-home-config</value>
+ </property>
+ <property>
+ <name>dt.test.3</name>
+ <value>user-home-config</value>
+ </property>
+ <property>
+ <name>dt.test.4</name>
+ <value>user-home-config</value>
+ </property>
+</configuration>
[32/36] incubator-apex-core git commit: Remove stray Copyright line.
Posted by vr...@apache.org.
Remove stray Copyright line.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/a190e86e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/a190e86e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/a190e86e
Branch: refs/heads/feature-module
Commit: a190e86e8f508308af539f5b473996e5fd4d6834
Parents: 6251310
Author: Thomas Weise <th...@datatorrent.com>
Authored: Thu Oct 22 21:14:05 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 23 12:45:08 2015 -0700
----------------------------------------------------------------------
.../resources/archetype-resources/XmlJavadocCommentsExtractor.xsl | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/a190e86e/apex-app-archetype/src/main/resources/archetype-resources/XmlJavadocCommentsExtractor.xsl
----------------------------------------------------------------------
diff --git a/apex-app-archetype/src/main/resources/archetype-resources/XmlJavadocCommentsExtractor.xsl b/apex-app-archetype/src/main/resources/archetype-resources/XmlJavadocCommentsExtractor.xsl
index 975ef1f..08075a9 100644
--- a/apex-app-archetype/src/main/resources/archetype-resources/XmlJavadocCommentsExtractor.xsl
+++ b/apex-app-archetype/src/main/resources/archetype-resources/XmlJavadocCommentsExtractor.xsl
@@ -1,8 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
- Copyright (C) 2015 DataTorrent, Inc.
-
Licensed 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
[06/36] incubator-apex-core git commit: APEX-162 Remove warnings and
redundant storage agent attributes from tests.
Posted by vr...@apache.org.
APEX-162 Remove warnings and redundant storage agent attributes from tests.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/265e9088
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/265e9088
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/265e9088
Branch: refs/heads/feature-module
Commit: 265e9088d2a3a4776cb8c7a1d2d1f35e561eb63b
Parents: 14f5269
Author: Thomas Weise <th...@datatorrent.com>
Authored: Tue Oct 13 19:05:47 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Tue Oct 13 19:05:47 2015 -0700
----------------------------------------------------------------------
.../test/java/com/datatorrent/stram/StramLocalClusterTest.java | 2 --
.../com/datatorrent/stram/StreamingContainerManagerTest.java | 6 ------
.../test/java/com/datatorrent/stram/engine/AutoMetricTest.java | 2 --
.../java/com/datatorrent/stram/engine/ProcessingModeTests.java | 4 ----
4 files changed, 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/265e9088/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
index aaf92b8..53d6f31 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
@@ -69,8 +69,6 @@ public class StramLocalClusterTest
@Test
public void testLocalClusterInitShutdown() throws Exception
{
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
-
TestGeneratorInputOperator genNode = dag.addOperator("genNode", TestGeneratorInputOperator.class);
genNode.setMaxTuples(2);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/265e9088/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
index b257632..295bcde 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
@@ -18,14 +18,12 @@
*/
package com.datatorrent.stram;
-import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import org.apache.commons.io.FileUtils;
import org.apache.hadoop.io.DataInputByteBuffer;
import org.apache.hadoop.io.DataOutputByteBuffer;
import org.junit.Assert;
@@ -35,7 +33,6 @@ import org.junit.Test;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
-import com.datatorrent.api.Context;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.Context.PortContext;
import com.datatorrent.api.DAG.Locality;
@@ -796,7 +793,6 @@ public class StreamingContainerManagerTest
@Test
public void testPhysicalPropertyUpdate() throws Exception
{
- dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
dag.addStream("o1.outport", o1.outport, o2.inport1);
@@ -837,7 +833,6 @@ public class StreamingContainerManagerTest
private void testAppDataSources(boolean appendQIDToTopic) throws Exception
{
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
StramLocalCluster lc = new StramLocalCluster(dag);
lc.runAsync();
StreamingContainerManager dnmgr = lc.dnmgr;
@@ -910,7 +905,6 @@ public class StreamingContainerManagerTest
try {
server.start();
int port = server.getPort();
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
dag.addStream("o1.outport", o1.outport, o2.inport1);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/265e9088/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java b/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
index f6451c9..28e2e51 100644
--- a/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
@@ -194,14 +194,12 @@ public class AutoMetricTest
@SuppressWarnings("SleepWhileInLoop")
public void testMetricPropagation() throws Exception
{
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
TestOperator testOper = dag.addOperator("TestOperator", TestOperator.class);
TestStatsListener sl = new TestStatsListener();
dag.setAttribute(testOper, OperatorContext.STATS_LISTENERS, Lists.newArrayList((StatsListener)sl));
- //dag.setAttribute(testOper, OperatorContext.INITIAL_PARTITION_COUNT, 1);
GenericTestOperator collector = dag.addOperator("Collector", new GenericTestOperator());
dag.addStream("TestTuples", testOper.outport, collector.inport1).setLocality(Locality.CONTAINER_LOCAL);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/265e9088/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java b/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
index 6df4e94..fa36fec 100644
--- a/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
+++ b/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
@@ -18,7 +18,6 @@
*/
package com.datatorrent.stram.engine;
-import com.datatorrent.common.util.AsyncFSStorageAgent;
import com.datatorrent.common.util.BaseOperator;
import java.io.IOException;
@@ -88,7 +87,6 @@ public class ProcessingModeTests
dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
RecoverableInputOperator rip = dag.addOperator("LongGenerator", RecoverableInputOperator.class);
rip.setMaximumTuples(maxTuples);
rip.setSimulateFailure(true);
@@ -110,7 +108,6 @@ public class ProcessingModeTests
CollectorOperator.collection.clear();
CollectorOperator.duplicates.clear();
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
@@ -134,7 +131,6 @@ public class ProcessingModeTests
CollectorOperator.collection.clear();
CollectorOperator.duplicates.clear();
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
[19/36] incubator-apex-core git commit: fixing ide templates
Posted by vr...@apache.org.
fixing ide templates
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/15ff42d8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/15ff42d8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/15ff42d8
Branch: refs/heads/feature-module
Commit: 15ff42d8d0d7d99e47e34aab6633756f046d270c
Parents: 529597f
Author: Chandni Singh <cs...@apache.org>
Authored: Mon Oct 19 23:46:11 2015 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Mon Oct 19 23:47:49 2015 -0700
----------------------------------------------------------------------
.idea/codeStyleSettings.xml | 102 +++++++
misc/ide-templates/eclipse/README.md | 17 ++
.../eclipse/apex-importorder.importorder | 10 +
misc/ide-templates/eclipse/apex-style.xml | 295 +++++++++++++++++++
misc/ide-templates/intellij/README.md | 24 ++
misc/ide-templates/intellij/apex-style.jar | Bin 0 -> 1515 bytes
misc/ide-templates/netbeans/README.md | 14 +
misc/ide-templates/netbeans/apex-style.zip | Bin 0 -> 2278 bytes
pom.xml | 2 +
9 files changed, 464 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/.idea/codeStyleSettings.xml
----------------------------------------------------------------------
diff --git a/.idea/codeStyleSettings.xml b/.idea/codeStyleSettings.xml
new file mode 100644
index 0000000..d469013
--- /dev/null
+++ b/.idea/codeStyleSettings.xml
@@ -0,0 +1,102 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="ProjectCodeStyleSettingsManager">
+ <option name="PER_PROJECT_SETTINGS">
+ <value>
+ <option name="OTHER_INDENT_OPTIONS">
+ <value>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="2" />
+ <option name="TAB_SIZE" value="2" />
+ <option name="USE_TAB_CHARACTER" value="false" />
+ <option name="SMART_TABS" value="false" />
+ <option name="LABEL_INDENT_SIZE" value="0" />
+ <option name="LABEL_INDENT_ABSOLUTE" value="false" />
+ <option name="USE_RELATIVE_INDENTS" value="false" />
+ </value>
+ </option>
+ <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="99" />
+ <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="99" />
+ <option name="IMPORT_LAYOUT_TABLE">
+ <value>
+ <package name="java" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="javax" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="org" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="org.apache" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="com" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="com.datatorrent" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="" withSubpackages="true" static="true" />
+ </value>
+ </option>
+ <option name="JD_P_AT_EMPTY_LINES" value="false" />
+ <JSON>
+ <option name="ARRAY_WRAPPING" value="0" />
+ </JSON>
+ <JavaCodeStyleSettings>
+ <option name="CLASS_NAMES_IN_JAVADOC" value="3" />
+ </JavaCodeStyleSettings>
+ <XML>
+ <option name="XML_TEXT_WRAP" value="0" />
+ <option name="XML_LEGACY_SETTINGS_IMPORTED" value="true" />
+ </XML>
+ <codeStyleSettings language="Groovy">
+ <indentOptions>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="2" />
+ <option name="TAB_SIZE" value="2" />
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="HTML">
+ <option name="RIGHT_MARGIN" value="120" />
+ <indentOptions>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="2" />
+ <option name="TAB_SIZE" value="2" />
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="JAVA">
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1" />
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1" />
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1" />
+ <option name="CLASS_BRACE_STYLE" value="2" />
+ <option name="METHOD_BRACE_STYLE" value="2" />
+ <option name="SPACE_AFTER_TYPE_CAST" value="false" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="DOWHILE_BRACE_FORCE" value="3" />
+ <option name="WHILE_BRACE_FORCE" value="3" />
+ <option name="FOR_BRACE_FORCE" value="3" />
+ <indentOptions>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="2" />
+ <option name="TAB_SIZE" value="2" />
+ <option name="LABEL_INDENT_SIZE" value="2" />
+ <option name="LABEL_INDENT_ABSOLUTE" value="true" />
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="JSON">
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="2" />
+ <option name="TAB_SIZE" value="2" />
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="XML">
+ <option name="RIGHT_MARGIN" value="120" />
+ <indentOptions>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="2" />
+ <option name="TAB_SIZE" value="2" />
+ </indentOptions>
+ </codeStyleSettings>
+ </value>
+ </option>
+ <option name="USE_PER_PROJECT_SETTINGS" value="true" />
+ </component>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/eclipse/README.md
----------------------------------------------------------------------
diff --git a/misc/ide-templates/eclipse/README.md b/misc/ide-templates/eclipse/README.md
new file mode 100644
index 0000000..723c6ca
--- /dev/null
+++ b/misc/ide-templates/eclipse/README.md
@@ -0,0 +1,17 @@
+Importing APEX Code Style in Eclipse
+====================================
+For Eclipse there are 2 configuration files
+1. apex-style.xml : this contains Java formatter settings.
+2. apex-importorder.importorder : this specifies the accepted import order for Apex projects.
+
+##To import Apex format style
+1. On the main menu, choose Eclipse | Preferences.
+2. Go to Java | Code Style | Formatter and click on 'Import'.
+3. In the Import Profile dialog box that opens select the `apex-style.xml`.
+4. Apply the changes.
+
+##To import Apex import order
+1. On the main menu, choose Eclipse | Preferences.
+2. Go to Java | Code Style | Organize Imports and click on 'Import'.
+3. In the Load Import Order from File dialog box that opens select `apex-importorder.importorder`.
+4. Apply the changes.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/eclipse/apex-importorder.importorder
----------------------------------------------------------------------
diff --git a/misc/ide-templates/eclipse/apex-importorder.importorder b/misc/ide-templates/eclipse/apex-importorder.importorder
new file mode 100644
index 0000000..6411400
--- /dev/null
+++ b/misc/ide-templates/eclipse/apex-importorder.importorder
@@ -0,0 +1,10 @@
+#Organize Import Order
+#Mon Oct 05 04:19:07 PDT 2015
+6=
+5=com.datatorrent
+4=com
+3=org.apache
+2=org
+1=javax
+0=java
+7=\#
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/eclipse/apex-style.xml
----------------------------------------------------------------------
diff --git a/misc/ide-templates/eclipse/apex-style.xml b/misc/ide-templates/eclipse/apex-style.xml
new file mode 100644
index 0000000..f5d0da0
--- /dev/null
+++ b/misc/ide-templates/eclipse/apex-style.xml
@@ -0,0 +1,295 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<profiles version="12">
+<profile kind="CodeFormatterProfile" name="apex-style" version="12">
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_ellipsis" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_enum_declarations" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_annotation_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_at_in_annotation_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.new_lines_at_block_boundaries" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_constructor_declaration_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.insert_new_line_for_parameter" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_package" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_after_imports" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_while" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.insert_new_line_before_root_tags" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_annotation_type_member_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_method_declaration_throws" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_javadoc_comments" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indentation.size" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_postfix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_for_increments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_type_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_for_inits" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_anonymous_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_semicolon_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.disabling_tag" value="@formatter:off"/>
+<setting id="org.eclipse.jdt.core.formatter.continuation_indentation" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_enum_constants" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_imports" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_after_package" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_binary_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_multiple_local_declarations" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_enum_constant" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_angle_bracket_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.indent_root_tags" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_before_or_operator_multicatch" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.enabling_tag" value="@formatter:on"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_brace_in_block" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_parenthesized_expression_in_return" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_throws_clause_in_method_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_parameter" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_then_statement_on_same_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_field" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_explicitconstructorcall_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_block" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_prefix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_between_type_declarations" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_brace_in_array_initializer" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_for" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_catch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_method" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_switch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_anonymous_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_parenthesized_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.never_indent_line_comments_on_first_column" value="false"/>
+<setting id="org.eclipse.jdt.core.compiler.problem.enumIdentifier" value="error"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_and_in_type_parameter" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_for_inits" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_statements_compare_to_block" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_anonymous_type_declaration" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_question_in_wildcard" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_method_invocation_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_switch" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.line_length" value="80"/>
+<setting id="org.eclipse.jdt.core.formatter.use_on_off_tags" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_brackets_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_enum_constant" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_assignment_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_for" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.preserve_white_space_between_code_and_line_comments" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_local_variable" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_method_declaration" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_union_type_in_multicatch" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.number_of_blank_lines_at_beginning_of_method_body" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_angle_bracket_in_type_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_else_statement_on_same_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_binary_expression" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_parameterized_type_reference" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_multiple_field_declarations" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_explicit_constructor_call" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_annotation_declaration_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_superinterfaces" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_default" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_question_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_block" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_constructor_declaration" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_lambda_body" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.compact_else_if" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_catch" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.put_empty_statement_on_new_line" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_parameters_in_constructor_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_method_invocation_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_method_invocation" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_throws_clause_in_constructor_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.compiler.problem.assertIdentifier" value="error"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.clear_blank_lines_in_block_comment" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_catch_in_try_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_try" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_at_end_of_file_if_missing" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.clear_blank_lines_in_javadoc_comment" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_array_initializer" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_binary_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_unary_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_expressions_in_array_initializer" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.format_line_comment_starting_on_first_column" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.number_of_empty_lines_to_preserve" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_case" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_ellipsis" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_semicolon_in_try_resources" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_assert" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_if" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_and_in_type_parameter" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_parenthesized_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_line_comments" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_labeled_statement" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.align_type_members_on_columns" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_assignment" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_method_body" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_type_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superinterfaces_in_type_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_first_class_body_declaration" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_conditional_expression" value="80"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_closing_brace_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_constructor_declaration_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.format_guardian_clause_on_one_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_if" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_type" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_block" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_enum_declaration" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_block_in_case" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_header" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_allocation_expression" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_while" value="insert"/>
+<setting id="org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode" value="enabled"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_switch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_method_declaration" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.join_wrapped_lines" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_switchstatements_compare_to_cases" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_bracket_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_synchronized" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.new_lines_at_javadoc_boundaries" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_annotation_type_declaration" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_resources_in_try" value="80"/>
+<setting id="org.eclipse.jdt.core.formatter.use_tabs_only_for_leading_indentations" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_selector_in_method_invocation" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.never_indent_block_comments_on_first_column" value="false"/>
+<setting id="org.eclipse.jdt.core.compiler.source" value="1.8"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_synchronized" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_constructor_declaration_throws" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.tabulation.size" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_enum_constant" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_allocation_expression" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_bracket_in_array_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_source_code" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_array_initializer" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_try" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_semicolon_in_try_resources" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_field" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_at_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.continuation_indentation_for_array_initializer" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_question_in_wildcard" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_method" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superclass_in_type_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superinterfaces_in_enum_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_parenthesized_expression_in_throw" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_labeled_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.compiler.codegen.targetPlatform" value="1.8"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_switch" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_superinterfaces" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_method_declaration_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_type_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_brace_in_array_initializer" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_parenthesized_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_html" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_at_in_annotation_type_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_angle_bracket_in_type_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_compact_if" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_empty_lines" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_unary_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_annotation" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_enum_declarations" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_empty_array_initializer_on_one_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_switchstatements_compare_to_switch" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_else_in_if_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_assignment_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_new_chunk" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_label" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_enum_declaration_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_bracket_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_angle_bracket_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_method_declaration_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_cast" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_assert" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_member_type" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_while_in_do_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_bracket_in_array_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_angle_bracket_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_qualified_allocation_expression" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_opening_brace_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_enum_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_breaks_compare_to_cases" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_if" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_semicolon" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_postfix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_try" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_cast" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_block_comments" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_lambda_arrow" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_imple_if_on_one_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_enum_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_parameters_in_method_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_brackets_in_array_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_angle_bracket_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_semicolon_in_for" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_method_declaration_throws" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_bracket_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_statements_compare_to_body" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_multiple_fields" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_enum_constant_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_prefix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_array_initializer" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_before_binary_operator" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_method_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_type_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_catch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.compiler.compliance" value="1.8"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_bracket_in_array_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_annotation" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_enum_constant_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_braces_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_case" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_multiple_local_declarations" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_annotation_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_bracket_in_array_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_outer_expressions_when_nested" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_paren_in_cast" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_enum_constant" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_type_declaration" value="next_line"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_package" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_synchronized" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_for_increments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_annotation_type_member_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_while" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_explicitconstructorcall_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_angle_bracket_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_enum_constant_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_lambda_arrow" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_constructor_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_constructor_declaration_throws" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.join_lines_in_comments" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_angle_bracket_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_question_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.indent_parameter_description" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_finally_in_try_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.tabulation.char" value="space"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_multiple_field_declarations" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_between_import_groups" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.lineSplit" value="120"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_switch" value="insert"/>
+</profile>
+</profiles>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/intellij/README.md
----------------------------------------------------------------------
diff --git a/misc/ide-templates/intellij/README.md b/misc/ide-templates/intellij/README.md
new file mode 100644
index 0000000..d65ab1d
--- /dev/null
+++ b/misc/ide-templates/intellij/README.md
@@ -0,0 +1,24 @@
+Importing APEX Code Style in IntelliJ
+=====================================
+
+With IntelliJ you can either import settings globally for all the workspaces or apply the style to a specific project.
+
+##Import settings globally
+To share settings between all the workspaces you can import the `apex-style.jar` which contains the apex-style.
+
+####To import the settings from the jar archive
+1. On the main menu, choose File | Import Settings.
+2. In the Import File Location dialog box that opens select the `apex-style.jar`.
+3. The Select Components to Import dialog box that opens next will only have an entry for code style which by default is selected. Click on OK.
+4. Restart IntelliJ once the jar is successfully imported.
+
+####To apply APEX code style
+1. On the main menu, choose IntelliJ | Preferences.
+2. Go to Editor | Code Style.
+3. In the 'Scheme' drop down select 'apex-style'.
+
+##Apply settings to a specific project
+1. There is a .idea folder under the Apex project which contains a codeStyleSettings.xml that describes APEX style.
+2. On the main menu of IntelliJ, choose IntelliJ | Preferences.
+3. Go to Editor | Code Style.
+4. In the 'Scheme' drop down select 'Project'.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/intellij/apex-style.jar
----------------------------------------------------------------------
diff --git a/misc/ide-templates/intellij/apex-style.jar b/misc/ide-templates/intellij/apex-style.jar
new file mode 100644
index 0000000..dc71b06
Binary files /dev/null and b/misc/ide-templates/intellij/apex-style.jar differ
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/netbeans/README.md
----------------------------------------------------------------------
diff --git a/misc/ide-templates/netbeans/README.md b/misc/ide-templates/netbeans/README.md
new file mode 100644
index 0000000..f2dbfcb
--- /dev/null
+++ b/misc/ide-templates/netbeans/README.md
@@ -0,0 +1,14 @@
+Importing APEX Code Style in NetBeans
+=====================================
+NetBeans supports importing settings bundled in a zip.
+This directory contains apex-style.zip which contains Apex format and import order configurations.
+
+##To import Apex code style
+1. On the main menu, choose NetBeans | Preferences.
+2. In the Options dialog, click on Editor | Formatting.
+3. Select the language as Java in the Language dropdown.
+4. Export the current settings for backup.
+5. Click on the 'Import' button which is at the bottom-left corner.
+6. In the Select Options To Import dialog box that opens select the `apex-style.zip` as the import source.
+7. Only formatting options are included which will be selected by default so click on OK.
+8. In the Import Confirmation dialog click on OK which will restart NetBeans.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/misc/ide-templates/netbeans/apex-style.zip
----------------------------------------------------------------------
diff --git a/misc/ide-templates/netbeans/apex-style.zip b/misc/ide-templates/netbeans/apex-style.zip
new file mode 100644
index 0000000..791cc0a
Binary files /dev/null and b/misc/ide-templates/netbeans/apex-style.zip differ
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/15ff42d8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 528eab4..13d5f52 100644
--- a/pom.xml
+++ b/pom.xml
@@ -112,6 +112,7 @@
<exclude>**/*.txt</exclude>
<exclude>**/*.importorder</exclude>
<exclude>**/archetype-resources/**</exclude>
+ <exclude>misc/ide-templates/**</exclude>
</excludes>
<mapping>
<dtcli>SCRIPT_STYLE</dtcli>
@@ -190,6 +191,7 @@
<exclude>**/*.md</exclude>
<exclude>**/*.txt</exclude>
<exclude>**/*.importorder</exclude>
+ <exclude>misc/ide-templates/**</exclude>
</excludes>
</configuration>
</plugin>
[25/36] incubator-apex-core git commit: Fix version lookup.
Posted by vr...@apache.org.
Fix version lookup.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/e3cecc26
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/e3cecc26
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/e3cecc26
Branch: refs/heads/feature-module
Commit: e3cecc2664e760931837c8caa399beb3518795de
Parents: 807e470
Author: Thomas Weise <th...@datatorrent.com>
Authored: Wed Oct 21 12:55:58 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Wed Oct 21 12:57:10 2015 -0700
----------------------------------------------------------------------
engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/e3cecc26/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java b/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
index fc29b01..ce083b5 100644
--- a/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
+++ b/engine/src/main/java/com/datatorrent/stram/util/VersionInfo.java
@@ -57,7 +57,7 @@ public class VersionInfo {
}
}
- Enumeration<URL> resources = VersionInfo.class.getClassLoader().getResources("META-INF/maven/com.datatorrent/dt-engine/pom.properties");
+ Enumeration<URL> resources = VersionInfo.class.getClassLoader().getResources("META-INF/maven/org.apache.apex/apex-engine/pom.properties");
while (resources.hasMoreElements()) {
Properties pomInfo = new Properties();
pomInfo.load(resources.nextElement().openStream());
[12/36] incubator-apex-core git commit: APEX-190 Replace Maven group
and artifact ID, configure ASF repositories, use apache-release profile.
Posted by vr...@apache.org.
APEX-190 Replace Maven group and artifact ID, configure ASF repositories, use apache-release profile.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/b1e7ea48
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/b1e7ea48
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/b1e7ea48
Branch: refs/heads/feature-module
Commit: b1e7ea48311e375f4487a7276d982e120fa6a5ab
Parents: 722fd67
Author: Thomas Weise <th...@datatorrent.com>
Authored: Tue Oct 13 17:15:27 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 16 13:03:50 2015 -0700
----------------------------------------------------------------------
apex-app-archetype/README.md | 2 +-
apex-app-archetype/pom.xml | 8 +--
.../main/resources/archetype-resources/pom.xml | 25 ++++---
apex-conf-archetype/README.md | 2 +-
apex-conf-archetype/pom.xml | 8 +--
api/pom.xml | 12 ++--
bufferserver/pom.xml | 14 ++--
common/pom.xml | 12 ++--
engine/pom.xml | 12 ++--
engine/src/main/scripts/dtcli | 2 +-
.../stram/client/AppPackageTest.java | 2 +-
.../resources/testAppPackage/mydtapp/pom.xml | 6 +-
pom.xml | 74 ++++++--------------
13 files changed, 77 insertions(+), 102 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/apex-app-archetype/README.md
----------------------------------------------------------------------
diff --git a/apex-app-archetype/README.md b/apex-app-archetype/README.md
index 775f026..9c386df 100644
--- a/apex-app-archetype/README.md
+++ b/apex-app-archetype/README.md
@@ -6,7 +6,7 @@ How to Generate an Apex Application Project Template
Run the following command
- mvn archetype:generate -DarchetypeGroupId=com.datatorrent -DarchetypeArtifactId=apex-app-archetype -DarchetypeVersion=3.0.0 -DgroupId=com.example -Dpackage=com.example.myapexapp -DartifactId=myapexapp -Dversion=1.0-SNAPSHOT
+ mvn archetype:generate -DarchetypeGroupId=org.apache.apex -DarchetypeArtifactId=apex-app-archetype -DarchetypeVersion=3.3.0-incubating-SNAPSHOT -DgroupId=com.example -Dpackage=com.example.myapexapp -DartifactId=myapexapp -Dversion=1.0-SNAPSHOT
Using your favorite IDE, open the project that has just been created by the above command.
Write your application code and optionally operator code
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/apex-app-archetype/pom.xml
----------------------------------------------------------------------
diff --git a/apex-app-archetype/pom.xml b/apex-app-archetype/pom.xml
index a26d36a..d9095cd 100644
--- a/apex-app-archetype/pom.xml
+++ b/apex-app-archetype/pom.xml
@@ -23,15 +23,15 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
</parent>
<artifactId>apex-app-archetype</artifactId>
<packaging>maven-archetype</packaging>
- <name>Apex Application Maven archetype</name>
+ <name>Apache Apex Application Maven Archetype</name>
<build>
<extensions>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/apex-app-archetype/src/main/resources/archetype-resources/pom.xml
----------------------------------------------------------------------
diff --git a/apex-app-archetype/src/main/resources/archetype-resources/pom.xml b/apex-app-archetype/src/main/resources/archetype-resources/pom.xml
index 33d1d1b..d3d0353 100644
--- a/apex-app-archetype/src/main/resources/archetype-resources/pom.xml
+++ b/apex-app-archetype/src/main/resources/archetype-resources/pom.xml
@@ -13,7 +13,7 @@
<properties>
<!-- change this if you desire to use a different version of DataTorrent -->
- <datatorrent.version>${archetypeVersion}</datatorrent.version>
+ <apex.version>${archetypeVersion}</apex.version>
<datatorrent.apppackage.classpath>lib/*.jar</datatorrent.apppackage.classpath>
</properties>
@@ -99,7 +99,7 @@
<archive>
<manifestEntries>
<Class-Path>${datatorrent.apppackage.classpath}</Class-Path>
- <DT-Engine-Version>${datatorrent.version}</DT-Engine-Version>
+ <DT-Engine-Version>${apex.version}</DT-Engine-Version>
<DT-App-Package-Name>${project.artifactId}</DT-App-Package-Name>
<DT-App-Package-Version>${project.version}</DT-App-Package-Version>
<DT-App-Package-Display-Name>${project.name}</DT-App-Package-Display-Name>
@@ -256,7 +256,14 @@
<dependency>
<groupId>com.datatorrent</groupId>
<artifactId>malhar-library</artifactId>
- <version>${datatorrent.version}</version>
+ <version>3.1.1</version>
+ <!-- remove when switching to malhar version that depends on apex-common as shown below -->
+ <exclusions>
+ <exclusion>
+ <groupId>com.datatorrent</groupId>
+ <artifactId>dt-common</artifactId>
+ </exclusion>
+ </exclusions>
<!--
If you know that your application does not need transitive dependencies pulled in by malhar-library,
uncomment the following to reduce the size of your app package.
@@ -271,9 +278,9 @@
-->
</dependency>
<dependency>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-common</artifactId>
- <version>${datatorrent.version}</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex-common</artifactId>
+ <version>${apex.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
@@ -283,9 +290,9 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-engine</artifactId>
- <version>${datatorrent.version}</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex-engine</artifactId>
+ <version>${apex.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/apex-conf-archetype/README.md
----------------------------------------------------------------------
diff --git a/apex-conf-archetype/README.md b/apex-conf-archetype/README.md
index d54876c..e80d1a2 100644
--- a/apex-conf-archetype/README.md
+++ b/apex-conf-archetype/README.md
@@ -6,7 +6,7 @@ How to Generate a Apex App Configuration Project Template
Run the following command
- mvn archetype:generate -DarchetypeGroupId=com.datatorrent -DarchetypeArtifactId=apex-conf-archetype -DarchetypeVersion=3.0.0 -DgroupId=com.example -Dpackage=com.example.myapexapp -DartifactId=myapexconf -Dversion=1.0-SNAPSHOT
+ mvn archetype:generate -DarchetypeGroupId=org.apache.apex -DarchetypeArtifactId=apex-conf-archetype -DarchetypeVersion=3.3.0-incubating-SNAPSHOT -DgroupId=com.example -Dpackage=com.example.myapexapp -DartifactId=myapexconf -Dversion=1.0-SNAPSHOT
Using your favorite IDE, open the project that has just been created by the above command.
Write your application code and optionally operator code
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/apex-conf-archetype/pom.xml
----------------------------------------------------------------------
diff --git a/apex-conf-archetype/pom.xml b/apex-conf-archetype/pom.xml
index 41501a8..4af8d16 100644
--- a/apex-conf-archetype/pom.xml
+++ b/apex-conf-archetype/pom.xml
@@ -23,15 +23,15 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
</parent>
<artifactId>apex-conf-archetype</artifactId>
<packaging>maven-archetype</packaging>
- <name>Apex App Configuration Maven archetype</name>
+ <name>Apache Apex App Configuration Maven Archetype</name>
<build>
<extensions>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/api/pom.xml
----------------------------------------------------------------------
diff --git a/api/pom.xml b/api/pom.xml
index 4262e8d..bb851bd 100644
--- a/api/pom.xml
+++ b/api/pom.xml
@@ -24,13 +24,13 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
</parent>
- <artifactId>dt-api</artifactId>
- <name>DataTorrent API</name>
+ <artifactId>apex-api</artifactId>
+ <name>Apache Apex API</name>
<packaging>jar</packaging>
<repositories>
@@ -139,7 +139,7 @@
</exclusions>
</dependency>
<dependency>
- <groupId>${project.groupId}</groupId>
+ <groupId>com.datatorrent</groupId>
<artifactId>netlet</artifactId>
<version>1.2.0</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/bufferserver/pom.xml
----------------------------------------------------------------------
diff --git a/bufferserver/pom.xml b/bufferserver/pom.xml
index 939da4f..a751ae9 100644
--- a/bufferserver/pom.xml
+++ b/bufferserver/pom.xml
@@ -23,15 +23,15 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
</parent>
- <artifactId>dt-bufferserver</artifactId>
+ <artifactId>apex-bufferserver</artifactId>
<description>Interface for nodes from 2 different containers to talk to each other.</description>
- <name>Buffer Server</name>
+ <name>Apache Apex Buffer Server</name>
<build>
<plugins>
@@ -65,8 +65,8 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-common</artifactId>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apex-common</artifactId>
<version>${project.version}</version>
<type>jar</type>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/common/pom.xml
----------------------------------------------------------------------
diff --git a/common/pom.xml b/common/pom.xml
index 898b45c..b6929ab 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -23,13 +23,13 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
</parent>
- <artifactId>dt-common</artifactId>
- <name>Base Library</name>
+ <artifactId>apex-common</artifactId>
+ <name>Apache Apex Common Library</name>
<packaging>jar</packaging>
<build>
@@ -83,7 +83,7 @@
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>dt-api</artifactId>
+ <artifactId>apex-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/engine/pom.xml
----------------------------------------------------------------------
diff --git a/engine/pom.xml b/engine/pom.xml
index 2165500..0c99ce0 100644
--- a/engine/pom.xml
+++ b/engine/pom.xml
@@ -23,15 +23,15 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
</parent>
- <artifactId>dt-engine</artifactId>
+ <artifactId>apex-engine</artifactId>
<packaging>jar</packaging>
- <name>Streaming Application Manager</name>
+ <name>Apache Apex Stream Processing Engine</name>
<build>
<finalName>${project.artifactId}</finalName>
@@ -165,7 +165,7 @@
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>dt-bufferserver</artifactId>
+ <artifactId>apex-bufferserver</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/engine/src/main/scripts/dtcli
----------------------------------------------------------------------
diff --git a/engine/src/main/scripts/dtcli b/engine/src/main/scripts/dtcli
index d7f9d76..e254215 100755
--- a/engine/src/main/scripts/dtcli
+++ b/engine/src/main/scripts/dtcli
@@ -67,7 +67,7 @@ fi
if [ -f "$MVN_GENERATED_PATH" ]; then
# development launch mode
- DT_CORE_JAR="$BUILD_DIR/dt-engine.jar"
+ DT_CORE_JAR="$BUILD_DIR/apex-engine.jar"
if [ ! -f "$DT_CORE_JAR" ]; then
echoerr "Error: Cannot find $DT_CORE_JAR";
exit 1;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java b/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
index 1224115..5a6c417 100644
--- a/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
@@ -84,7 +84,7 @@ public class AppPackageTest
{
Assert.assertEquals("mydtapp", json.getString("appPackageName"));
Assert.assertEquals("1.0-SNAPSHOT", json.getString("appPackageVersion"));
- Assert.assertEquals("3.0.0", json.getString("dtEngineVersion"));
+ Assert.assertEquals("3.3.0-incubating-SNAPSHOT", json.getString("dtEngineVersion"));
Assert.assertEquals("lib/*.jar", json.getJSONArray("classPath").getString(0));
JSONObject application = json.getJSONArray("applications").getJSONObject(0);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/engine/src/test/resources/testAppPackage/mydtapp/pom.xml
----------------------------------------------------------------------
diff --git a/engine/src/test/resources/testAppPackage/mydtapp/pom.xml b/engine/src/test/resources/testAppPackage/mydtapp/pom.xml
index abbca75..d8f61e4 100644
--- a/engine/src/test/resources/testAppPackage/mydtapp/pom.xml
+++ b/engine/src/test/resources/testAppPackage/mydtapp/pom.xml
@@ -33,7 +33,7 @@
<properties>
<!-- change this if you desire to use a different version of DataTorrent -->
- <datatorrent.version>3.0.0</datatorrent.version>
+ <datatorrent.version>3.3.0-incubating-SNAPSHOT</datatorrent.version>
<datatorrent.apppackage.classpath>lib/*.jar</datatorrent.apppackage.classpath>
</properties>
@@ -184,8 +184,8 @@
<dependencies>
<!-- add your dependencies here -->
<dependency>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-common</artifactId>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex-common</artifactId>
<version>${datatorrent.version}</version>
<scope>provided</scope>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b1e7ea48/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index e75d3c3..7f41873 100644
--- a/pom.xml
+++ b/pom.xml
@@ -22,12 +22,18 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
- <groupId>com.datatorrent</groupId>
- <artifactId>dt-framework</artifactId>
- <version>3.3.0-SNAPSHOT</version>
+ <parent>
+ <groupId>org.apache</groupId>
+ <artifactId>apache</artifactId>
+ <version>16</version>
+ </parent>
+
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex</artifactId>
+ <version>3.3.0-incubating-SNAPSHOT</version>
<packaging>pom</packaging>
- <name>Apache Apex Core</name>
+ <name>Apache Apex</name>
<url>http://apex.apache.org</url>
<organization>
@@ -43,6 +49,7 @@
</licenses>
<repositories>
+ <!-- added for semantic versioning check, won't be needed once we publish to central -->
<repository>
<snapshots>
<enabled>false</enabled>
@@ -63,7 +70,6 @@
<jersey.version>1.9</jersey.version>
<!-- do not change jetty version as later versions have problems with DefaultServlet -->
<jetty.version>8.1.10.v20130312</jetty.version>
- <maven.deploy.repo.classifier></maven.deploy.repo.classifier>
<license.skip>true</license.skip>
<checkstyle.console>false</checkstyle.console>
</properties>
@@ -340,30 +346,29 @@
</build>
<profiles>
<profile>
- <id>release</id>
+ <id>apache-release</id>
<modules>
</modules>
<properties>
<maven.compiler.debug>false</maven.compiler.debug>
<maven.compiler.optimize>true</maven.compiler.optimize>
<maven.deploy.skip>false</maven.deploy.skip>
- <!-- always generate on release build -->
- <generate-revision-info-phase>generate-resources</generate-revision-info-phase>
- <distMgmtDevUrl>internal.repo::default::file://${project.build.directory}/mvn-repo</distMgmtDevUrl>
- <distMgmtUrl>dav:http://www.datatorrent.com:8081/nexus/content/repositories</distMgmtUrl>
+ <distMgmtSnapshotsId>apache.snapshots.https</distMgmtSnapshotsId>
+ <distMgmtSnapshotsUrl>https://repository.apache.org/content/repositories/snapshots</distMgmtSnapshotsUrl>
+ <distMgmtReleaseId>apache.staging.https</distMgmtReleaseId>
+ <distMgmtReleaseUrl>https://repository.apache.org/service/local/staging/deploy/maven2</distMgmtReleaseUrl>
+ <distMgmtDevUrl>file://${project.build.directory}/mvn-repo</distMgmtDevUrl>
<package.prefix>/opt/datatorrent</package.prefix>
<package.groupname>dtorrent</package.groupname>
</properties>
<distributionManagement>
<repository>
- <id>releases</id>
- <name>DataTorrent Repository</name>
- <url>${distMgmtUrl}/${maven.deploy.repo.classifier}releases/</url>
+ <id>${distMgmtReleaseId}</id>
+ <url>${distMgmtReleaseUrl}</url>
</repository>
<snapshotRepository>
- <id>snapshots</id>
- <name>DataTorrent Snapshots</name>
- <url>${distMgmtUrl}/${maven.deploy.repo.classifier}snapshots/</url>
+ <id>${distMgmtSnapshotsId}</id>
+ <url>${distMgmtSnapshotsUrl}</url>
</snapshotRepository>
</distributionManagement>
<build>
@@ -381,43 +386,6 @@
</execution>
</executions>
</plugin>
- <plugin>
- <artifactId>maven-site-plugin</artifactId>
- <version>3.2</version>
- </plugin>
- <plugin>
- <artifactId>maven-deploy-plugin</artifactId>
- <version>2.7</version>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-javadoc-plugin</artifactId>
- <executions>
- <execution>
- <!-- build javadoc jars per jar for publishing to maven -->
- <id>module-javadocs</id>
- <phase>prepare-package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-source-plugin</artifactId>
- <version>2.2.1</version>
- <executions>
- <execution>
- <id>attach-sources</id>
- <goals>
- <goal>jar-no-fork</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
</plugins>
</build>
</profile>
[16/36] incubator-apex-core git commit: Merge branch 'APEX-197' of
https://github.com/chandnisingh/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'APEX-197' of https://github.com/chandnisingh/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/dac8fa8f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/dac8fa8f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/dac8fa8f
Branch: refs/heads/feature-module
Commit: dac8fa8f73b53533937fea1d3d34b4cb496f3e67
Parents: ccee82f 330a284
Author: Thomas Weise <th...@datatorrent.com>
Authored: Sat Oct 17 23:11:56 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Oct 17 23:11:56 2015 -0700
----------------------------------------------------------------------
pom.xml | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/dac8fa8f/pom.xml
----------------------------------------------------------------------
[13/36] incubator-apex-core git commit: Merge branch
'APEX-137.update-contrib-section' of
https://github.com/andyperlitch/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'APEX-137.update-contrib-section' of https://github.com/andyperlitch/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/f48074ae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/f48074ae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/f48074ae
Branch: refs/heads/feature-module
Commit: f48074ae04de0e7e84e54faa4eed1a9f245cf6e0
Parents: b1e7ea4 b336f64
Author: Thomas Weise <th...@datatorrent.com>
Authored: Fri Oct 16 14:14:51 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 16 14:14:51 2015 -0700
----------------------------------------------------------------------
README.md | 15 +--------------
1 file changed, 1 insertion(+), 14 deletions(-)
----------------------------------------------------------------------
[03/36] incubator-apex-core git commit: APEX-193 #resolve remove
useless entry from archetype-metadata
Posted by vr...@apache.org.
APEX-193 #resolve remove useless entry from archetype-metadata
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/87673779
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/87673779
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/87673779
Branch: refs/heads/feature-module
Commit: 8767377927d60ab8e3d483043170f7376bd81c57
Parents: 809e6f6
Author: David Yan <da...@datatorrent.com>
Authored: Mon Oct 12 17:53:58 2015 -0700
Committer: David Yan <da...@datatorrent.com>
Committed: Mon Oct 12 17:53:58 2015 -0700
----------------------------------------------------------------------
.../src/main/resources/META-INF/maven/archetype-metadata.xml | 6 ------
1 file changed, 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/87673779/apex-app-archetype/src/main/resources/META-INF/maven/archetype-metadata.xml
----------------------------------------------------------------------
diff --git a/apex-app-archetype/src/main/resources/META-INF/maven/archetype-metadata.xml b/apex-app-archetype/src/main/resources/META-INF/maven/archetype-metadata.xml
index c93a8d9..98951f8 100644
--- a/apex-app-archetype/src/main/resources/META-INF/maven/archetype-metadata.xml
+++ b/apex-app-archetype/src/main/resources/META-INF/maven/archetype-metadata.xml
@@ -35,12 +35,6 @@
</includes>
</fileSet>
<fileSet filtered="true" encoding="UTF-8">
- <directory>src/main/resources</directory>
- <includes>
- <include>**/*.xml</include>
- </includes>
- </fileSet>
- <fileSet filtered="true" encoding="UTF-8">
<directory>src/main/resources/META-INF</directory>
</fileSet>
<fileSet filtered="true" encoding="UTF-8">
[24/36] incubator-apex-core git commit: Merge branch 'APEX-184' of
github.com:vrozov/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'APEX-184' of github.com:vrozov/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/807e470e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/807e470e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/807e470e
Branch: refs/heads/feature-module
Commit: 807e470e8b2d25c00981c7ce24fd7a1fd945fb96
Parents: 42dd9f8 5b6e428
Author: Pramod Immaneni <pr...@datatorrent.com>
Authored: Wed Oct 21 05:41:29 2015 -0700
Committer: Pramod Immaneni <pr...@datatorrent.com>
Committed: Wed Oct 21 05:41:29 2015 -0700
----------------------------------------------------------------------
bufferserver/pom.xml | 2 +-
.../bufferserver/internal/DataList.java | 248 ++++++++++++-------
.../bufferserver/internal/DataListener.java | 2 +-
.../bufferserver/internal/FastDataList.java | 11 +-
.../bufferserver/internal/LogicalNode.java | 37 ++-
.../datatorrent/bufferserver/server/Server.java | 108 ++++----
6 files changed, 232 insertions(+), 176 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/807e470e/bufferserver/pom.xml
----------------------------------------------------------------------
[17/36] incubator-apex-core git commit: Merge branch 'SPOI-5053' of
https://github.com/ishark/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'SPOI-5053' of https://github.com/ishark/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/d7afe728
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/d7afe728
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/d7afe728
Branch: refs/heads/feature-module
Commit: d7afe728091ff6d795dda202d0a3579d1ede88a9
Parents: dac8fa8 fb33c23
Author: Thomas Weise <th...@datatorrent.com>
Authored: Sat Oct 17 23:28:21 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Oct 17 23:28:21 2015 -0700
----------------------------------------------------------------------
.../stram/StreamingContainerManager.java | 35 +++++++++++---------
.../stram/StreamingContainerManagerTest.java | 11 ++----
2 files changed, 22 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/d7afe728/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
----------------------------------------------------------------------
[09/36] incubator-apex-core git commit: Fix JavaDoc in
AsyncFSStorageAgent.
Posted by vr...@apache.org.
Fix JavaDoc in AsyncFSStorageAgent.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/722fd672
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/722fd672
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/722fd672
Branch: refs/heads/feature-module
Commit: 722fd6720f44af166cc1bcbd54e1cd8fed0b993f
Parents: ecb19fa
Author: Thomas Weise <th...@datatorrent.com>
Authored: Thu Oct 15 11:02:56 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Thu Oct 15 11:05:37 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/common/util/AsyncFSStorageAgent.java | 10 ++++++----
1 file changed, 6 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/722fd672/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java b/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
index 0e23c2e..83bbdca 100644
--- a/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
+++ b/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
@@ -16,10 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-/**
- * @since 3.1.0
- */
-
package com.datatorrent.common.util;
import java.io.*;
@@ -33,6 +29,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * <p>AsyncFSStorageAgent class.</p>
+ *
+ * @since 3.1.0
+ */
public class AsyncFSStorageAgent extends FSStorageAgent
{
private final transient Configuration conf;
[02/36] incubator-apex-core git commit: APEX-162 #resolve Enhance
StramTestSupport.TestMeta API.
Posted by vr...@apache.org.
APEX-162 #resolve Enhance StramTestSupport.TestMeta API.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/3c35cccb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/3c35cccb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/3c35cccb
Branch: refs/heads/feature-module
Commit: 3c35cccbd574d2a28e543b644deeb9f7c8a886e5
Parents: 809e6f6
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Mon Oct 12 16:39:32 2015 -0700
Committer: Vlad Rozov <v....@datatorrent.com>
Committed: Mon Oct 12 16:39:32 2015 -0700
----------------------------------------------------------------------
engine/pom.xml | 2 +-
.../com/datatorrent/stram/CheckpointTest.java | 55 ++++--------
.../stram/LogicalPlanModificationTest.java | 34 ++++----
.../datatorrent/stram/OutputUnifiedTest.java | 29 ++++---
.../stram/StramLocalClusterTest.java | 17 ++--
.../datatorrent/stram/StramMiniClusterTest.java | 6 +-
.../datatorrent/stram/StramRecoveryTest.java | 48 +++++------
.../com/datatorrent/stram/StreamCodecTest.java | 51 ++---------
.../stram/StreamingContainerManagerTest.java | 90 +++++++-------------
.../stram/engine/AutoMetricTest.java | 22 ++---
.../stram/engine/ProcessingModeTests.java | 17 ++--
.../stram/plan/StreamPersistanceTests.java | 27 ++----
.../stram/support/StramTestSupport.java | 64 ++++++++++++--
13 files changed, 212 insertions(+), 250 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/pom.xml
----------------------------------------------------------------------
diff --git a/engine/pom.xml b/engine/pom.xml
index 7974313..2165500 100644
--- a/engine/pom.xml
+++ b/engine/pom.xml
@@ -145,7 +145,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>2322</maxAllowedViolations>
+ <maxAllowedViolations>2320</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/CheckpointTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/CheckpointTest.java b/engine/src/test/java/com/datatorrent/stram/CheckpointTest.java
index ae28ebd..5d11b86 100644
--- a/engine/src/test/java/com/datatorrent/stram/CheckpointTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/CheckpointTest.java
@@ -19,8 +19,7 @@
package com.datatorrent.stram;
import com.datatorrent.common.util.BaseOperator;
-import java.io.File;
-import java.io.IOException;
+
import java.util.*;
import com.google.common.collect.Maps;
@@ -30,8 +29,6 @@ import org.junit.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.hadoop.fs.FileContext;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.util.SystemClock;
@@ -51,6 +48,7 @@ import com.datatorrent.stram.plan.logical.LogicalPlan;
import com.datatorrent.stram.plan.physical.PTContainer;
import com.datatorrent.stram.plan.physical.PTOperator;
import com.datatorrent.stram.plan.physical.PhysicalPlan;
+import com.datatorrent.stram.support.StramTestSupport;
import com.datatorrent.stram.support.StramTestSupport.MemoryStorageAgent;
import com.datatorrent.stram.support.StramTestSupport.TestMeta;
@@ -61,30 +59,9 @@ public class CheckpointTest
{
@SuppressWarnings("unused")
private static final Logger LOG = LoggerFactory.getLogger(CheckpointTest.class);
- @Rule public TestMeta testMeta = new TestMeta();
-
- /**
- *
- * @throws IOException
- */
- @Before
- public void setupEachTest() throws IOException
- {
- try {
- FileContext.getLocalFSFileContext().delete(
- new Path(new File(testMeta.dir).getAbsolutePath()), true);
- }
- catch (Exception e) {
- throw new RuntimeException("could not cleanup test dir", e);
- }
- //StramChild.eventloop.start();
- }
- @After
- public void teardown()
- {
- //StramChild.eventloop.stop();
- }
+ @Rule
+ public TestMeta testMeta = new TestMeta();
private static class MockInputOperator extends BaseOperator implements InputOperator
{
@@ -106,6 +83,14 @@ public class CheckpointTest
}
}
+ private LogicalPlan dag;
+
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
+ }
+
/**
* Test saving of operator state at window boundary.
* @throws Exception
@@ -113,9 +98,7 @@ public class CheckpointTest
@Test
public void testBackup() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
- AsyncFSStorageAgent storageAgent = new AsyncFSStorageAgent(testMeta.dir, null);
+ AsyncFSStorageAgent storageAgent = new AsyncFSStorageAgent(testMeta.getPath(), null);
storageAgent.setSyncCheckpoint(true);
dag.setAttribute(OperatorContext.STORAGE_AGENT, storageAgent);
dag.setAttribute(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 1);
@@ -172,8 +155,7 @@ public class CheckpointTest
public void testUpdateRecoveryCheckpoint() throws Exception
{
Clock clock = new SystemClock();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+
dag.setAttribute(com.datatorrent.api.Context.OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
@@ -273,8 +255,7 @@ public class CheckpointTest
public void testUpdateCheckpointsRecovery()
{
MockClock clock = new MockClock();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+
dag.setAttribute(com.datatorrent.api.Context.OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
dag.setAttribute(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 1);
@@ -338,8 +319,7 @@ public class CheckpointTest
public void testUpdateCheckpointsProcessingTimeout()
{
MockClock clock = new MockClock();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+
dag.setAttribute(com.datatorrent.api.Context.OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
@@ -419,8 +399,7 @@ public class CheckpointTest
public void testBlockedOperatorContainerRestart()
{
MockClock clock = new MockClock();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+
dag.setAttribute(com.datatorrent.api.Context.OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java b/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
index 847f3fd..8a50124 100644
--- a/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
@@ -26,6 +26,7 @@ import java.util.concurrent.FutureTask;
import javax.validation.ValidationException;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -46,18 +47,26 @@ import com.datatorrent.stram.plan.physical.PTContainer;
import com.datatorrent.stram.plan.physical.PTOperator;
import com.datatorrent.stram.plan.physical.PhysicalPlan;
import com.datatorrent.stram.plan.physical.PlanModifier;
+import com.datatorrent.stram.support.StramTestSupport;
import com.datatorrent.stram.support.StramTestSupport.TestMeta;
import com.google.common.collect.Sets;
public class LogicalPlanModificationTest
{
- @Rule public TestMeta testMeta = new TestMeta();
+ private LogicalPlan dag;
+
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
+ }
@Test
public void testAddOperator()
{
- LogicalPlan dag = new LogicalPlan();
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class);
@@ -94,7 +103,6 @@ public class LogicalPlanModificationTest
@Test
public void testSetOperatorProperty()
{
- LogicalPlan dag = new LogicalPlan();
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
OperatorMeta o1Meta = dag.getMeta(o1);
@@ -121,8 +129,6 @@ public class LogicalPlanModificationTest
@Test
public void testRemoveOperator()
{
- LogicalPlan dag = new LogicalPlan();
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
OperatorMeta o1Meta = dag.getMeta(o1);
GenericTestOperator o12 = dag.addOperator("o12", GenericTestOperator.class);
@@ -191,8 +197,6 @@ public class LogicalPlanModificationTest
@Test
public void testRemoveOperator2()
{
- LogicalPlan dag = new LogicalPlan();
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
OperatorMeta o1Meta = dag.getMeta(o1);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
@@ -233,8 +237,6 @@ public class LogicalPlanModificationTest
@Test
public void testRemoveStream()
{
- LogicalPlan dag = new LogicalPlan();
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
@@ -255,8 +257,6 @@ public class LogicalPlanModificationTest
@Test
public void testAddStream()
{
- LogicalPlan dag = new LogicalPlan();
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
@@ -295,10 +295,8 @@ public class LogicalPlanModificationTest
}
- private void testExecutionManager(StorageAgent agent) throws Exception {
-
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
+ private void testExecutionManager(StorageAgent agent) throws Exception
+ {
dag.setAttribute(OperatorContext.STORAGE_AGENT, agent);
StreamingContainerManager dnm = new StreamingContainerManager(dag);
@@ -337,13 +335,13 @@ public class LogicalPlanModificationTest
@Test
public void testExecutionManagerWithSyncStorageAgent() throws Exception
{
- testExecutionManager(new FSStorageAgent(testMeta.dir, null));
+ testExecutionManager(new FSStorageAgent(testMeta.getPath(), null));
}
@Test
public void testExecutionManagerWithAsyncStorageAgent() throws Exception
{
- testExecutionManager(new AsyncFSStorageAgent(testMeta.dir, null));
+ testExecutionManager(new AsyncFSStorageAgent(testMeta.getPath(), null));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java b/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
index f0461ba..7581cc3 100644
--- a/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
@@ -22,6 +22,7 @@ import java.net.InetSocketAddress;
import java.util.List;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -50,12 +51,18 @@ public class OutputUnifiedTest
@Rule
public StramTestSupport.TestMeta testMeta = new StramTestSupport.TestMeta();
- @Test
- public void testManyToOnePartition() throws Exception {
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
+ private LogicalPlan dag;
+
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
+ }
+ @Test
+ public void testManyToOnePartition() throws Exception
+ {
TestInputOperator i1 = new TestInputOperator();
dag.addOperator("i1", i1);
@@ -83,11 +90,8 @@ public class OutputUnifiedTest
}
@Test
- public void testMxNPartition() throws Exception {
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
-
+ public void testMxNPartition() throws Exception
+ {
TestInputOperator i1 = new TestInputOperator();
dag.addOperator("i1", i1);
@@ -117,11 +121,8 @@ public class OutputUnifiedTest
}
@Test
- public void testParallelPartition() throws Exception {
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
-
+ public void testParallelPartition() throws Exception
+ {
TestInputOperator i1 = new TestInputOperator();
dag.addOperator("i1", i1);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
index c784fd1..aaf92b8 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
@@ -29,8 +29,6 @@ import org.junit.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.datatorrent.api.Context;
-
import com.datatorrent.common.util.AsyncFSStorageAgent;
import com.datatorrent.stram.StramLocalCluster.LocalStreamingContainer;
import com.datatorrent.stram.StramLocalCluster.MockComponentFactory;
@@ -48,11 +46,12 @@ public class StramLocalClusterTest
@Rule
public StramTestSupport.TestMeta testMeta = new StramTestSupport.TestMeta();
+ private LogicalPlan dag;
+
@Before
public void setup() throws IOException
{
-// StramChild.eventloop = new DefaultEventLoop("StramLocalClusterTestEventLoop");
-// StramChild.eventloop.start();
+ dag = StramTestSupport.createDAG(testMeta);
}
@After
@@ -70,9 +69,7 @@ public class StramLocalClusterTest
@Test
public void testLocalClusterInitShutdown() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
- dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
TestGeneratorInputOperator genNode = dag.addOperator("genNode", TestGeneratorInputOperator.class);
genNode.setMaxTuples(2);
@@ -110,11 +107,9 @@ public class StramLocalClusterTest
@SuppressWarnings("SleepWhileInLoop")
public void testRecovery() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
- AsyncFSStorageAgent agent = new AsyncFSStorageAgent(testMeta.dir, null);
+ AsyncFSStorageAgent agent = new AsyncFSStorageAgent(testMeta.getPath(), null);
agent.setSyncCheckpoint(true);
- dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, agent);
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, agent);
TestGeneratorInputOperator node1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
// data will be added externally from test
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java b/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
index f0fd325..d5cb14f 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
@@ -227,8 +227,8 @@ public class StramMiniClusterTest
private LogicalPlan createDAG(LogicalPlanConfiguration lpc) throws Exception
{
LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, new File(testMeta.dir).toURI().toString());
- lpc.prepareDAG(dag,null,"testApp");
+ dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.toURI().toString());
+ lpc.prepareDAG(dag, null, "testApp");
dag.validate();
Assert.assertEquals("", Integer.valueOf(128), dag.getValue(DAG.MASTER_MEMORY_MB));
Assert.assertEquals("", "-Dlog4j.properties=custom_log4j.properties", dag.getValue(DAG.CONTAINER_JVM_OPTIONS));
@@ -360,7 +360,7 @@ public class StramMiniClusterTest
{
LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, new File(testMeta.dir).toURI().toString());
+ dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.toURI().toString());
FailingOperator badOperator = dag.addOperator("badOperator", FailingOperator.class);
dag.getContextAttributes(badOperator).put(OperatorContext.RECOVERY_ATTEMPTS, 1);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java b/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
index 89ae3e7..6dbdcf0 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.ipc.RPC.Server;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.test.MockitoUtil;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.Mockito;
@@ -65,6 +66,7 @@ import com.datatorrent.stram.plan.physical.PTContainer;
import com.datatorrent.stram.plan.physical.PTOperator;
import com.datatorrent.stram.plan.physical.PhysicalPlan;
import com.datatorrent.stram.plan.physical.PhysicalPlanTest.PartitioningTestOperator;
+import com.datatorrent.stram.support.StramTestSupport;
import com.datatorrent.stram.support.StramTestSupport.TestMeta;
import static org.junit.Assert.assertEquals;
@@ -72,12 +74,20 @@ import static org.junit.Assert.assertEquals;
public class StramRecoveryTest
{
private static final Logger LOG = LoggerFactory.getLogger(StramRecoveryTest.class);
- @Rule public final TestMeta testMeta = new TestMeta();
- private void testPhysicalPlanSerialization(StorageAgent agent) throws Exception
+ @Rule
+ public final TestMeta testMeta = new TestMeta();
+
+ private LogicalPlan dag;
+
+ @Before
+ public void setup()
{
- LogicalPlan dag = new LogicalPlan();
+ dag = StramTestSupport.createDAG(testMeta);
+ }
+ private void testPhysicalPlanSerialization(StorageAgent agent) throws Exception
+ {
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
PartitioningTestOperator o2 = dag.addOperator("o2", PartitioningTestOperator.class);
o2.setPartitionCount(3);
@@ -127,13 +137,13 @@ public class StramRecoveryTest
@Test
public void testPhysicalPlanSerializationWithSyncAgent() throws Exception
{
- testPhysicalPlanSerialization(new FSStorageAgent(testMeta.dir, null));
+ testPhysicalPlanSerialization(new FSStorageAgent(testMeta.getPath(), null));
}
@Test
public void testPhysicalPlanSerializationWithAsyncAgent() throws Exception
{
- testPhysicalPlanSerialization(new AsyncFSStorageAgent(testMeta.dir, null));
+ testPhysicalPlanSerialization(new AsyncFSStorageAgent(testMeta.getPath(), null));
}
public static class StatsListeningOperator extends TestGeneratorInputOperator implements StatsListener
@@ -161,10 +171,6 @@ public class StramRecoveryTest
*/
private void testContainerManager(StorageAgent agent) throws Exception
{
- FileUtils.deleteDirectory(new File(testMeta.dir)); // clean any state from previous run
-
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
dag.setAttribute(OperatorContext.STORAGE_AGENT, agent);
StatsListeningOperator o1 = dag.addOperator("o1", StatsListeningOperator.class);
@@ -188,8 +194,7 @@ public class StramRecoveryTest
assertEquals("state " + o1p1, PTOperator.State.PENDING_DEPLOY, o1p1.getState());
// test restore initial snapshot + log
- dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+ dag = StramTestSupport.createDAG(testMeta);
scm = StreamingContainerManager.getInstance(new FSRecoveryHandler(dag.assertAppPath(), new Configuration(false)), dag, false);
dag = scm.getLogicalPlan();
plan = scm.getPhysicalPlan();
@@ -245,8 +250,7 @@ public class StramRecoveryTest
checkpoint(scm, o1p1, offlineCheckpoint);
// test restore
- dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+ dag = StramTestSupport.createDAG(testMeta);
scm = StreamingContainerManager.getInstance(new FSRecoveryHandler(dag.assertAppPath(), new Configuration(false)), dag, false);
Assert.assertNotSame("dag references", dag, scm.getLogicalPlan());
@@ -270,13 +274,13 @@ public class StramRecoveryTest
@Test
public void testContainerManagerWithSyncAgent() throws Exception
{
- testPhysicalPlanSerialization(new FSStorageAgent(testMeta.dir, null));
+ testPhysicalPlanSerialization(new FSStorageAgent(testMeta.getPath(), null));
}
@Test
public void testContainerManagerWithAsyncAgent() throws Exception
{
- testPhysicalPlanSerialization(new AsyncFSStorageAgent(testMeta.dir, null));
+ testPhysicalPlanSerialization(new AsyncFSStorageAgent(testMeta.getPath(), null));
}
@Test
@@ -284,9 +288,7 @@ public class StramRecoveryTest
{
final MutableInt flushCount = new MutableInt();
final MutableBoolean isClosed = new MutableBoolean(false);
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new FSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new FSStorageAgent(testMeta.getPath(), null));
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
StreamingContainerManager scm = new StreamingContainerManager(dag);
@@ -386,12 +388,10 @@ public class StramRecoveryTest
private void testRestartApp(StorageAgent agent, String appPath1) throws Exception
{
- FileUtils.deleteDirectory(new File(testMeta.dir)); // clean any state from previous run
String appId1 = "app1";
String appId2 = "app2";
- String appPath2 = testMeta.dir + "/" + appId2;
+ String appPath2 = testMeta.getPath() + "/" + appId2;
- LogicalPlan dag = new LogicalPlan();
dag.setAttribute(LogicalPlan.APPLICATION_ID, appId1);
dag.setAttribute(LogicalPlan.APPLICATION_PATH, appPath1);
dag.setAttribute(LogicalPlan.APPLICATION_ATTEMPT_ID, 1);
@@ -445,21 +445,21 @@ public class StramRecoveryTest
@Test
public void testRestartAppWithSyncAgent() throws Exception
{
- String appPath1 = testMeta.dir + "/app1";
+ final String appPath1 = testMeta.getPath() + "/app1";
testRestartApp(new FSStorageAgent(appPath1 + "/" + LogicalPlan.SUBDIR_CHECKPOINTS, null), appPath1);
}
@Test
public void testRestartAppWithAsyncAgent() throws Exception
{
- String appPath1 = testMeta.dir + "/app1";
+ final String appPath1 = testMeta.getPath() + "/app1";
testRestartApp(new AsyncFSStorageAgent(appPath1 + "/" + LogicalPlan.SUBDIR_CHECKPOINTS, null), appPath1);
}
@Test
public void testRpcFailover() throws Exception
{
- String appPath = testMeta.dir;
+ String appPath = testMeta.getPath();
Configuration conf = new Configuration(false);
final AtomicBoolean timedout = new AtomicBoolean();
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java b/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
index 6bfa591..ddf3448 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
@@ -34,6 +34,7 @@ import com.google.common.collect.Lists;
import java.io.Serializable;
import java.util.*;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -41,15 +42,20 @@ import org.junit.Test;
*/
public class StreamCodecTest
{
+ private LogicalPlan dag;
+
@Rule
public StramTestSupport.TestMeta testMeta = new StramTestSupport.TestMeta();
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
+ }
+
@Test
public void testStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -107,9 +113,6 @@ public class StreamCodecTest
@Test
public void testStreamCodecReuse()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -154,9 +157,6 @@ public class StreamCodecTest
@Test
public void testDefaultStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
DefaultCodecOperator node2 = dag.addOperator("node2", DefaultCodecOperator.class);
DefaultCodecOperator node3 = dag.addOperator("node3", DefaultCodecOperator.class);
@@ -213,9 +213,6 @@ public class StreamCodecTest
@Test
public void testPartitioningStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
dag.setAttribute(node2, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(3));
@@ -264,9 +261,6 @@ public class StreamCodecTest
@Test
public void testMxNPartitioningStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
dag.setAttribute(node1, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
@@ -331,9 +325,6 @@ public class StreamCodecTest
@Test
public void testParallelPartitioningStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
dag.setAttribute(node1, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
@@ -422,9 +413,6 @@ public class StreamCodecTest
@Test
public void testMultipleInputStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
TestStreamCodec serDe = new TestStreamCodec();
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
@@ -477,9 +465,6 @@ public class StreamCodecTest
@Test
public void testPartitioningMultipleInputStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
dag.setAttribute(node1, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
@@ -556,9 +541,6 @@ public class StreamCodecTest
@Test
public void testMultipleStreamCodecs()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -613,9 +595,6 @@ public class StreamCodecTest
@Test
public void testPartitioningMultipleStreamCodecs()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -699,9 +678,6 @@ public class StreamCodecTest
@Test
public void testMxNMultipleStreamCodecs()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
dag.setAttribute(node1, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
@@ -803,9 +779,6 @@ public class StreamCodecTest
@Test
public void testInlineStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -887,9 +860,6 @@ public class StreamCodecTest
@Test
public void testCascadingStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -974,9 +944,6 @@ public class StreamCodecTest
@Test
public void testDynamicPartitioningStreamCodec()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
dag.setAttribute(node1, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
dag.setAttribute(node1, Context.OperatorContext.STATS_LISTENERS, Lists.newArrayList((StatsListener) new PartitioningTest.PartitionLoadWatch()));
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
index 710440d..b257632 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
@@ -29,6 +29,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.hadoop.io.DataInputByteBuffer;
import org.apache.hadoop.io.DataOutputByteBuffer;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -83,10 +84,20 @@ import org.eclipse.jetty.websocket.WebSocket;
public class StreamingContainerManagerTest
{
- @Rule public TestMeta testMeta = new TestMeta();
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ private LogicalPlan dag;
+
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
+ }
@Test
- public void testDeployInfoSerialization() throws Exception {
+ public void testDeployInfoSerialization() throws Exception
+ {
OperatorDeployInfo ndi = new OperatorDeployInfo();
ndi.name = "node1";
ndi.type = OperatorDeployInfo.OperatorType.GENERIC;
@@ -136,11 +147,8 @@ public class StreamingContainerManagerTest
}
@Test
- public void testGenerateDeployInfo() {
-
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
+ public void testGenerateDeployInfo()
+ {
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class);
@@ -238,9 +246,8 @@ public class StreamingContainerManagerTest
}
@Test
- public void testStaticPartitioning() {
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
+ public void testStaticPartitioning()
+ {
//
// ,---> node2----,
// | |
@@ -355,9 +362,6 @@ public class StreamingContainerManagerTest
@Test
public void testRecoveryOrder() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
GenericTestOperator node3 = dag.addOperator("node3", GenericTestOperator.class);
@@ -406,9 +410,6 @@ public class StreamingContainerManagerTest
@Test
public void testRecoveryUpstreamInline() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class);
@@ -444,11 +445,9 @@ public class StreamingContainerManagerTest
}
@Test
- public void testCheckpointWindowIds() throws Exception {
- File path = new File(testMeta.dir);
- FileUtils.deleteDirectory(path.getAbsoluteFile());
-
- FSStorageAgent sa = new FSStorageAgent(path.getPath(), null);
+ public void testCheckpointWindowIds() throws Exception
+ {
+ FSStorageAgent sa = new FSStorageAgent(testMeta.getPath(), null);
long[] windowIds = new long[]{123L, 345L, 234L};
for (long windowId : windowIds) {
@@ -475,11 +474,7 @@ public class StreamingContainerManagerTest
@Test
public void testAsyncCheckpointWindowIds() throws Exception
{
- File path = new File(testMeta.dir);
- FileUtils.deleteDirectory(path.getAbsoluteFile());
- FileUtils.forceMkdir(new File(path.getAbsoluteFile(), "/localPath"));
-
- AsyncFSStorageAgent sa = new AsyncFSStorageAgent(path.getPath(), null);
+ AsyncFSStorageAgent sa = new AsyncFSStorageAgent(testMeta.getPath(), null);
long[] windowIds = new long[]{123L, 345L, 234L};
for (long windowId : windowIds) {
@@ -506,13 +501,8 @@ public class StreamingContainerManagerTest
@Test
public void testProcessHeartbeat() throws Exception
{
- FileUtils.deleteDirectory(new File(testMeta.dir)); // clean any state from previous run
-
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
-
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
- dag.setAttribute(o1, OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{new PartitioningTest.PartitionLoadWatch()}));
+ dag.setAttribute(o1, OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{new PartitioningTest.PartitionLoadWatch()}));
dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
StreamingContainerManager scm = new StreamingContainerManager(dag);
@@ -655,9 +645,6 @@ public class StreamingContainerManagerTest
@Test
public void testValidGenericOperatorDeployInfoType()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
TestGeneratorInputOperator.ValidGenericOperator o2 = dag.addOperator("o2", TestGeneratorInputOperator.ValidGenericOperator.class);
@@ -683,9 +670,6 @@ public class StreamingContainerManagerTest
@Test
public void testValidInputOperatorDeployInfoType()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
TestGeneratorInputOperator.ValidInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.ValidInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
@@ -711,8 +695,6 @@ public class StreamingContainerManagerTest
@Test
public void testOperatorShutdown()
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class);
@@ -781,7 +763,6 @@ public class StreamingContainerManagerTest
private void testDownStreamPartition(Locality locality) throws Exception
{
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
dag.setAttribute(o2, OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
@@ -815,8 +796,7 @@ public class StreamingContainerManagerTest
@Test
public void testPhysicalPropertyUpdate() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
dag.addStream("o1.outport", o1.outport, o2.inport1);
@@ -837,10 +817,9 @@ public class StreamingContainerManagerTest
lc.shutdown();
}
- private LogicalPlan getTestAppDataSourceLogicalPlan(Class<? extends TestAppDataQueryOperator> qClass,
+ private void setupAppDataSourceLogicalPlan(Class<? extends TestAppDataQueryOperator> qClass,
Class<? extends TestAppDataSourceOperator> dsClass, Class<? extends TestAppDataResultOperator> rClass)
{
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
TestAppDataQueryOperator q = dag.addOperator("q", qClass);
TestAppDataResultOperator r = dag.addOperator("r", rClass);
@@ -854,13 +833,11 @@ public class StreamingContainerManagerTest
dag.addStream("o1-to-ds", o1.outport, ds.inport1);
dag.addStream("q-to-ds", q.outport, ds.query);
dag.addStream("ds-to-r", ds.result, r.inport);
-
- return dag;
}
- private void testAppDataSources(LogicalPlan dag, boolean appendQIDToTopic) throws Exception
+ private void testAppDataSources(boolean appendQIDToTopic) throws Exception
{
- dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
StramLocalCluster lc = new StramLocalCluster(dag);
lc.runAsync();
StreamingContainerManager dnmgr = lc.dnmgr;
@@ -883,22 +860,22 @@ public class StreamingContainerManagerTest
@Test
public void testGetAppDataSources1() throws Exception
{
- LogicalPlan dag = getTestAppDataSourceLogicalPlan(TestAppDataQueryOperator.class, TestAppDataSourceOperator.class, TestAppDataResultOperator.ResultOperator1.class);
- testAppDataSources(dag, true);
+ setupAppDataSourceLogicalPlan(TestAppDataQueryOperator.class, TestAppDataSourceOperator.class, TestAppDataResultOperator.ResultOperator1.class);
+ testAppDataSources(true);
}
@Test
public void testGetAppDataSources2() throws Exception
{
- LogicalPlan dag = getTestAppDataSourceLogicalPlan(TestAppDataQueryOperator.class, TestAppDataSourceOperator.class, TestAppDataResultOperator.ResultOperator2.class);
- testAppDataSources(dag, false);
+ setupAppDataSourceLogicalPlan(TestAppDataQueryOperator.class, TestAppDataSourceOperator.class, TestAppDataResultOperator.ResultOperator2.class);
+ testAppDataSources(false);
}
@Test
public void testGetAppDataSources3() throws Exception
{
- LogicalPlan dag = getTestAppDataSourceLogicalPlan(TestAppDataQueryOperator.class, TestAppDataSourceOperator.class, TestAppDataResultOperator.ResultOperator3.class);
- testAppDataSources(dag, false);
+ setupAppDataSourceLogicalPlan(TestAppDataQueryOperator.class, TestAppDataSourceOperator.class, TestAppDataResultOperator.ResultOperator3.class);
+ testAppDataSources(false);
}
@Test
@@ -933,8 +910,7 @@ public class StreamingContainerManagerTest
try {
server.start();
int port = server.getPort();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
dag.addStream("o1.outport", o1.outport, o2.inport1);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java b/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
index a1312a5..f6451c9 100644
--- a/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/engine/AutoMetricTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.CountDownLatch;
import org.apache.hadoop.conf.Configuration;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
@@ -176,6 +177,14 @@ public class AutoMetricTest
}
+ private LogicalPlan dag;
+
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
+ }
+
/**
* Verify custom stats generated by operator are propagated and trigger repartition.
*
@@ -185,8 +194,7 @@ public class AutoMetricTest
@SuppressWarnings("SleepWhileInLoop")
public void testMetricPropagation() throws Exception
{
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
@@ -234,9 +242,7 @@ public class AutoMetricTest
CountDownLatch latch = new CountDownLatch(1);
LogicalPlanConfiguration lpc = new LogicalPlanConfiguration(new Configuration());
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
TestGeneratorInputOperator inputOperator = dag.addOperator("input", TestGeneratorInputOperator.class);
OperatorWithMetrics o1 = dag.addOperator("o1", OperatorWithMetrics.class);
@@ -264,8 +270,7 @@ public class AutoMetricTest
CountDownLatch latch = new CountDownLatch(2);
LogicalPlanConfiguration lpc = new LogicalPlanConfiguration(new Configuration());
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+
TestGeneratorInputOperator inputOperator = dag.addOperator("input", TestGeneratorInputOperator.class);
OperatorWithMetrics o1 = dag.addOperator("o1", OperatorWithMetrics.class);
@@ -289,8 +294,7 @@ public class AutoMetricTest
public void testInjectionOfDefaultMetricsAggregator() throws Exception
{
LogicalPlanConfiguration lpc = new LogicalPlanConfiguration(new Configuration());
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
+
TestGeneratorInputOperator inputOperator = dag.addOperator("input", TestGeneratorInputOperator.class);
OperatorWithMetricMethod o1 = dag.addOperator("o1", OperatorWithMetricMethod.class);
@@ -365,9 +369,7 @@ public class AutoMetricTest
CountDownLatch latch = new CountDownLatch(1);
LogicalPlanConfiguration lpc = new LogicalPlanConfiguration(new Configuration());
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.dir);
TestGeneratorInputOperator inputOperator = dag.addOperator("input", TestGeneratorInputOperator.class);
OperatorWithMetricMethod o1 = dag.addOperator("o1", OperatorWithMetricMethod.class);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java b/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
index cef671e..6df4e94 100644
--- a/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
+++ b/engine/src/test/java/com/datatorrent/stram/engine/ProcessingModeTests.java
@@ -43,6 +43,7 @@ import com.datatorrent.bufferserver.packet.MessageType;
import com.datatorrent.bufferserver.util.Codec;
import com.datatorrent.stram.StramLocalCluster;
import com.datatorrent.stram.plan.logical.LogicalPlan;
+import com.datatorrent.stram.support.StramTestSupport;
import com.datatorrent.stram.support.StramTestSupport.TestMeta;
import com.datatorrent.stram.tuple.EndWindowTuple;
import com.datatorrent.stram.tuple.Tuple;
@@ -52,7 +53,11 @@ import com.datatorrent.stram.tuple.Tuple;
*/
public class ProcessingModeTests
{
- @Rule public TestMeta testMeta = new TestMeta();
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ private LogicalPlan dag;
+
ProcessingMode processingMode;
int maxTuples = 30;
@@ -64,6 +69,7 @@ public class ProcessingModeTests
@Before
public void setup() throws IOException
{
+ dag = StramTestSupport.createDAG(testMeta);
StreamingContainer.eventloop.start();
}
@@ -79,11 +85,10 @@ public class ProcessingModeTests
CollectorOperator.collection.clear();
CollectorOperator.duplicates.clear();
- LogicalPlan dag = new LogicalPlan();
dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
RecoverableInputOperator rip = dag.addOperator("LongGenerator", RecoverableInputOperator.class);
rip.setMaximumTuples(maxTuples);
rip.setSimulateFailure(true);
@@ -105,8 +110,7 @@ public class ProcessingModeTests
CollectorOperator.collection.clear();
CollectorOperator.duplicates.clear();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
@@ -130,8 +134,7 @@ public class ProcessingModeTests
CollectorOperator.collection.clear();
CollectorOperator.duplicates.clear();
- LogicalPlan dag = new LogicalPlan();
- dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.dir, null));
+ dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(testMeta.getPath(), null));
dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/plan/StreamPersistanceTests.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/plan/StreamPersistanceTests.java b/engine/src/test/java/com/datatorrent/stram/plan/StreamPersistanceTests.java
index 1839c91..5dca8a4 100644
--- a/engine/src/test/java/com/datatorrent/stram/plan/StreamPersistanceTests.java
+++ b/engine/src/test/java/com/datatorrent/stram/plan/StreamPersistanceTests.java
@@ -31,6 +31,7 @@ import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.slf4j.Logger;
@@ -203,10 +204,17 @@ public class StreamPersistanceTests
{
}
+ private LogicalPlan dag;
+
+ @Before
+ public void setup()
+ {
+ dag = StramTestSupport.createDAG(testMeta);
+ }
+
@Test
public void testPersistStreamOperatorIsAdded()
{
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x = dag.addOperator("x", new GenericTestOperator());
TestRecieverOperator persister = new TestRecieverOperator();
@@ -222,7 +230,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOperatorIsAddedPerSink()
{
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x1 = dag.addOperator("x1", new GenericTestOperator());
GenericTestOperator x2 = dag.addOperator("x2", new GenericTestOperator());
@@ -255,7 +262,6 @@ public class StreamPersistanceTests
public void testaddStreamThrowsExceptionOnInvalidLoggerType()
{
// Test Logger with non-optional output ports
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x = dag.addOperator("x", new GenericTestOperator());
StreamMeta stream = dag.addStream("Stream1", input1.outport, x.inport1);
@@ -302,7 +308,6 @@ public class StreamPersistanceTests
public void testaddStreamThrowsExceptionOnInvalidInputPortForLoggerType()
{
// Test for input port belonging to different object
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x = dag.addOperator("x", new GenericTestOperator());
TestRecieverOperator persister = new TestRecieverOperator();
@@ -322,7 +327,6 @@ public class StreamPersistanceTests
public void testPersistStreamOperatorIsRemovedWhenStreamIsRemoved()
{
// Remove Stream and check if persist operator is removed
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x = dag.addOperator("x", new GenericTestOperator());
TestRecieverOperator persister = new TestRecieverOperator();
@@ -340,7 +344,6 @@ public class StreamPersistanceTests
public void testPersistStreamOperatorIsRemovedWhenSinkIsRemoved()
{
// Remove sink and check if corresponding persist operator is removed
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x1 = dag.addOperator("x1", new GenericTestOperator());
GenericTestOperator x2 = dag.addOperator("x2", new GenericTestOperator());
@@ -383,7 +386,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOperatorIsRemovedWhenAllSinksAreRemoved()
{
- LogicalPlan dag = new LogicalPlan();
TestGeneratorInputOperator input1 = dag.addOperator("input1", TestGeneratorInputOperator.class);
GenericTestOperator x1 = dag.addOperator("x1", new GenericTestOperator());
GenericTestOperator x2 = dag.addOperator("x2", new GenericTestOperator());
@@ -409,7 +411,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOperatorGeneratesIdenticalOutputAsSink() throws ClassNotFoundException, IOException, InterruptedException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator input1 = dag.addOperator("input1", AscendingNumbersOperator.class);
// Add PersistOperator directly to dag
final TestRecieverOperator x = dag.addOperator("x", new TestRecieverOperator());
@@ -603,7 +604,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamWithFiltering() throws ClassNotFoundException, IOException, InterruptedException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
PassThruOperatorWithCodec passThru = dag.addOperator("PassThrough", new PassThruOperatorWithCodec(2));
TestRecieverOperator console = dag.addOperator("console", new TestRecieverOperator());
@@ -617,7 +617,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOnSingleSinkWithFiltering() throws ClassNotFoundException, IOException, InterruptedException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
PassThruOperatorWithCodec passThru = dag.addOperator("PassThrough", new PassThruOperatorWithCodec(2));
final TestRecieverOperator console = dag.addOperator("console", new TestRecieverOperator());
@@ -632,7 +631,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOnSingleSinkWithFilteringContainerLocal() throws ClassNotFoundException, IOException, InterruptedException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
PassThruOperatorWithCodec passThru = dag.addOperator("PassThrough", new PassThruOperatorWithCodec(2));
PassThruOperatorWithCodec passThru2 = dag.addOperator("Multiples_of_3", new PassThruOperatorWithCodec(3));
@@ -696,7 +694,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOperatorGeneratesUnionOfAllSinksOutput() throws ClassNotFoundException, IOException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
PassThruOperatorWithCodec passThru1 = dag.addOperator("PassThrough1", new PassThruOperatorWithCodec(2));
PassThruOperatorWithCodec passThru2 = dag.addOperator("PassThrough2", new PassThruOperatorWithCodec(3));
@@ -829,7 +826,6 @@ public class StreamPersistanceTests
@Test
public void testPersistStreamOperatorMultiplePhysicalOperatorsForSink() throws ClassNotFoundException, IOException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
PartitionedTestOperatorWithFiltering passThru = dag.addOperator("partition", new PartitionedTestOperatorWithFiltering());
final TestRecieverOperator console = dag.addOperator("console", new TestRecieverOperator());
@@ -883,7 +879,6 @@ public class StreamPersistanceTests
@Test
public void testPartitionedPersistOperator() throws ClassNotFoundException, IOException
{
- LogicalPlan dag = new LogicalPlan();
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
PartitionedTestOperatorWithFiltering passThru = dag.addOperator("partition", new PartitionedTestOperatorWithFiltering());
final TestRecieverOperator console = dag.addOperator("console", new TestRecieverOperator());
@@ -942,10 +937,6 @@ public class StreamPersistanceTests
@Test
public void testDynamicPartitioning() throws ClassNotFoundException, IOException
{
- LogicalPlan dag = new LogicalPlan();
-
- dag.setAttribute(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, testMeta.dir);
-
AscendingNumbersOperator ascend = dag.addOperator("ascend", new AscendingNumbersOperator());
final TestRecieverOperator console = dag.addOperator("console", new TestRecieverOperator());
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3c35cccb/engine/src/test/java/com/datatorrent/stram/support/StramTestSupport.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/support/StramTestSupport.java b/engine/src/test/java/com/datatorrent/stram/support/StramTestSupport.java
index 314fdfc..cf2a887 100644
--- a/engine/src/test/java/com/datatorrent/stram/support/StramTestSupport.java
+++ b/engine/src/test/java/com/datatorrent/stram/support/StramTestSupport.java
@@ -22,6 +22,10 @@ import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
+import java.net.URI;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -50,6 +54,7 @@ import com.datatorrent.stram.api.AppDataSource;
import com.datatorrent.stram.api.BaseContext;
import com.datatorrent.stram.engine.OperatorContext;
import com.datatorrent.stram.engine.WindowGenerator;
+import com.datatorrent.stram.plan.logical.LogicalPlan;
import com.datatorrent.stram.plan.physical.PTOperator;
import com.datatorrent.stram.tuple.EndWindowTuple;
import com.datatorrent.stram.tuple.Tuple;
@@ -253,23 +258,54 @@ abstract public class StramTestSupport
public static class TestMeta extends TestWatcher
{
- public String dir = null;
+ private File dir;
@Override
protected void starting(org.junit.runner.Description description)
{
- String methodName = description.getMethodName();
- String className = description.getClassName();
- //className = className.substring(className.lastIndexOf('.') + 1);
- this.dir = "target/" + className + "/" + methodName;
- new File(this.dir).mkdirs();
+ final String methodName = description.getMethodName();
+ final String className = description.getClassName();
+ dir = new File("target/" + className + "/" + methodName);
+ try {
+ Files.createDirectories(dir.toPath());
+ } catch (FileAlreadyExistsException e) {
+ try {
+ Files.delete(dir.toPath());
+ Files.createDirectories(dir.toPath());
+ } catch (IOException ioe) {
+ throw new RuntimeException("Fail to create test working directory " + dir.getAbsolutePath(), e);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Fail to create test working directory " + dir.getAbsolutePath(), e);
+ }
}
@Override
protected void finished(org.junit.runner.Description description)
{
- FileUtils.deleteQuietly(new File(this.dir));
+ FileUtils.deleteQuietly(dir);
+ }
+
+ public String getPath()
+ {
+ return dir.getPath();
+ }
+
+ public String getAbsolutePath()
+ {
+ return dir.getAbsolutePath();
+ }
+
+ public Path toPath()
+ {
+ return dir.toPath();
+ }
+
+ public URI toURI()
+ {
+ return dir.toURI();
}
+
}
public static class TestHomeDirectory extends TestWatcher
@@ -333,6 +369,20 @@ abstract public class StramTestSupport
}
}
+ public static LogicalPlan createDAG(final TestMeta testMeta, final String suffix)
+ {
+ if (suffix == null) {
+ throw new NullPointerException();
+ }
+ LogicalPlan dag = new LogicalPlan();
+ dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.getPath() + suffix);
+ return dag;
+ }
+
+ public static LogicalPlan createDAG(final TestMeta testMeta)
+ {
+ return createDAG(testMeta, "");
+ }
public static class MemoryStorageAgent implements StorageAgent, Serializable
{
[07/36] incubator-apex-core git commit: Merge branch 'APEX-132' into
devel-3
Posted by vr...@apache.org.
Merge branch 'APEX-132' into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/ecb19fa7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/ecb19fa7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/ecb19fa7
Branch: refs/heads/feature-module
Commit: ecb19fa783c0d5eaecfd9056681557d3343a44e7
Parents: 265e908 5086212
Author: Chandni Singh <cs...@apache.org>
Authored: Wed Oct 14 15:35:07 2015 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Wed Oct 14 15:35:07 2015 -0700
----------------------------------------------------------------------
.idea/copyright/apache.xml | 6 ++++++
.idea/copyright/profiles_settings.xml | 9 +++++++++
2 files changed, 15 insertions(+)
----------------------------------------------------------------------
[33/36] incubator-apex-core git commit: Add (incubating) to NOTICE
Posted by vr...@apache.org.
Add (incubating) to NOTICE
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/40fba973
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/40fba973
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/40fba973
Branch: refs/heads/feature-module
Commit: 40fba973ea333c8c10d3d5fa0e2b6a876d917e6f
Parents: b44c646
Author: Thomas Weise <th...@datatorrent.com>
Authored: Thu Oct 22 22:22:27 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Fri Oct 23 12:46:22 2015 -0700
----------------------------------------------------------------------
NOTICE | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/40fba973/NOTICE
----------------------------------------------------------------------
diff --git a/NOTICE b/NOTICE
index 1dfedde..d0f6855 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,4 +1,4 @@
-Apache Apex
+Apache Apex (incubating)
Copyright (c) 2015 The Apache Software Foundation
This product includes software developed at
[15/36] incubator-apex-core git commit: Merge branch 'SPOI-6505' of
https://github.com/ishark/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'SPOI-6505' of https://github.com/ishark/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/ccee82fd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/ccee82fd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/ccee82fd
Branch: refs/heads/feature-module
Commit: ccee82fd0a2dc2c3a5dcd682705d38f392613fec
Parents: f48074a 7d0a644
Author: Thomas Weise <th...@datatorrent.com>
Authored: Sat Oct 17 20:02:05 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Oct 17 20:02:05 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/webapp/asm/CompactUtil.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
[04/36] incubator-apex-core git commit: SPOI-5053 APEX-56 #resolve
#comment Fixing removal of terminated operators from physical plan when
downanStream operators are also completed till shutdown window Id Also fixed
containers to be removed only when ope
Posted by vr...@apache.org.
SPOI-5053 APEX-56 #resolve #comment
Fixing removal of terminated operators from physical plan when downanStream operators are also completed till shutdown window Id
Also fixed containers to be removed only when operators are removed from physical plan
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/fb33c23e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/fb33c23e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/fb33c23e
Branch: refs/heads/feature-module
Commit: fb33c23e141c422c85eefc06e680e9ecbf459936
Parents: 809e6f6
Author: ishark <is...@datatorrent.com>
Authored: Mon Oct 12 16:05:32 2015 -0700
Committer: ishark <is...@datatorrent.com>
Committed: Tue Oct 13 18:17:03 2015 -0700
----------------------------------------------------------------------
.../stram/StreamingContainerManager.java | 35 +++++++++++---------
.../stram/StreamingContainerManagerTest.java | 11 ++----
2 files changed, 22 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/fb33c23e/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
index ed366db..3931fad 100644
--- a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
+++ b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
@@ -1040,10 +1040,7 @@ public class StreamingContainerManager implements PlanContext
Iterator<Map.Entry<Long, Set<PTOperator>>> it = shutdownOperators.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<Long, Set<PTOperator>> windowAndOpers = it.next();
- if (windowAndOpers.getKey().longValue() > this.committedWindowId) {
- // wait until window is committed
- continue;
- } else {
+ if (windowAndOpers.getKey().longValue() <= this.committedWindowId || checkDownStreamOperators(windowAndOpers)) {
LOG.info("Removing inactive operators at window {} {}", Codec.getStringWindowId(windowAndOpers.getKey()), windowAndOpers.getValue());
for (PTOperator oper : windowAndOpers.getValue()) {
plan.removeTerminatedPartition(oper);
@@ -1070,8 +1067,7 @@ public class StreamingContainerManager implements PlanContext
try {
command.run();
count++;
- }
- catch (Exception e) {
+ } catch (Exception e) {
// TODO: handle error
LOG.error("Failed to execute {}", command, e);
}
@@ -1081,8 +1077,7 @@ public class StreamingContainerManager implements PlanContext
if (count > 0) {
try {
checkpoint();
- }
- catch (Exception e) {
+ } catch (Exception e) {
throw new RuntimeException("Failed to checkpoint state.", e);
}
}
@@ -1090,6 +1085,19 @@ public class StreamingContainerManager implements PlanContext
return count;
}
+ private boolean checkDownStreamOperators(Map.Entry<Long, Set<PTOperator>> windowAndOpers)
+ {
+ // Check if all downStream operators are at higher window Ids, then operator can be removed from dag
+ Set<PTOperator> downStreamOperators = getPhysicalPlan().getDependents(windowAndOpers.getValue());
+ for (PTOperator oper : downStreamOperators) {
+ long windowId = oper.stats.currentWindowId.get();
+ if (windowId < windowAndOpers.getKey().longValue()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* Schedule container restart. Called by Stram after a container was terminated
* and requires recovery (killed externally, or after heartbeat timeout). <br>
@@ -1495,8 +1503,6 @@ public class StreamingContainerManager implements PlanContext
}
Set<Integer> reportedOperators = Sets.newHashSetWithExpectedSize(sca.container.getOperators().size());
- boolean containerIdle = true;
-
for (OperatorHeartbeat shb : heartbeat.getContainerStats().operators) {
long maxEndWindowTimestamp = 0;
@@ -1533,9 +1539,7 @@ public class StreamingContainerManager implements PlanContext
oper.stats.lastHeartbeat = shb;
List<ContainerStats.OperatorStats> statsList = shb.getOperatorStatsContainer();
- if (!oper.stats.isIdle()) {
- containerIdle = false;
- }
+
if (!statsList.isEmpty()) {
long tuplesProcessed = 0;
long tuplesEmitted = 0;
@@ -1743,11 +1747,10 @@ public class StreamingContainerManager implements PlanContext
ContainerHeartbeatResponse rsp = getHeartbeatResponse(sca);
- if (containerIdle && isApplicationIdle()) {
+ if (heartbeat.getContainerStats().operators.isEmpty() && isApplicationIdle()) {
LOG.info("requesting idle shutdown for container {}", heartbeat.getContainerId());
rsp.shutdown = true;
- }
- else {
+ } else {
if (sca.shutdownRequested) {
LOG.info("requesting shutdown for container {}", heartbeat.getContainerId());
rsp.shutdown = true;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/fb33c23e/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
index 710440d..2884323 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
@@ -762,18 +762,13 @@ public class StreamingContainerManagerTest
mc1.sendHeartbeat();
scm.monitorHeartbeat();
+ Assert.assertEquals("committedWindowId", 1, scm.getCommittedWindowId());
+
o2p1mos.currentWindowId(2).checkpointWindowId(2);
mc2.sendHeartbeat();
scm.monitorHeartbeat();
- Assert.assertEquals("committedWindowId", 1, scm.getCommittedWindowId());
- scm.monitorHeartbeat(); // committedWindowId updated in next cycle
- Assert.assertEquals("committedWindowId", 2, scm.getCommittedWindowId());
- Assert.assertEquals(1, o1p1.getContainer().getOperators().size());
- Assert.assertEquals(1, o2p1.getContainer().getOperators().size());
- Assert.assertEquals(2, physicalPlan.getContainers().size());
- // call again as events are processed after committed window was updated
- scm.processEvents();
+ // Operators are shutdown when both operators reach window Id 2
Assert.assertEquals(0, o1p1.getContainer().getOperators().size());
Assert.assertEquals(0, o2p1.getContainer().getOperators().size());
Assert.assertEquals(0, physicalPlan.getContainers().size());
[10/36] incubator-apex-core git commit: APEX-137 #resolve
Posted by vr...@apache.org.
APEX-137 #resolve
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/b336f64d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/b336f64d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/b336f64d
Branch: refs/heads/feature-module
Commit: b336f64df18907caab50957313cbd7270535c62d
Parents: 722fd67
Author: Andy Perlitch <an...@datatorrent.com>
Authored: Thu Oct 15 15:14:13 2015 -0700
Committer: Andy Perlitch <an...@datatorrent.com>
Committed: Thu Oct 15 15:14:13 2015 -0700
----------------------------------------------------------------------
README.md | 15 +--------------
1 file changed, 1 insertion(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/b336f64d/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index baabcd2..a4a2b46 100644
--- a/README.md
+++ b/README.md
@@ -13,22 +13,9 @@ Please visit the [documentation section](http://apex.incubator.apache.org/docs.h
##Contributing
-This project welcomes new contributors. If you would like to help by adding new features, enhancements or fixing bugs, here is how to do it.
+This project welcomes new contributors. If you would like to help by adding new features, enhancements or fixing bugs, check out the [contributing guidelines](http://apex.incubator.apache.org/contributing.html).
You acknowledge that your submissions to this repository are made pursuant the terms of the Apache License, Version 2.0 (http://www.apache.org/licenses/LICENSE-2.0.html) and constitute "Contributions," as defined therein, and you represent and warrant that you have the right and authority to do so.
-
- * Fork your own GitHub repository
- * Create a topic branch with an appropriate name
- * Write code, comments, tests in your repository
- * Create a GitHub pull request from your repository, providing as many details about your changes as possible
- * After review and acceptance one of the committers will merge the pull request.
-
-When adding **new files**, please include the Apache v2.0 license header. From the top level directory:
-
-Run `mvn license:check -Dlicense.skip=false` to check correct header formatting.
-Run `mvn license:format -Dlicense.skip=false` to automatically add the header when missing.
-
-Thanks for contributing!
##Building Apex
[14/36] incubator-apex-core git commit: Addressing review comments
Posted by vr...@apache.org.
Addressing review comments
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/7d0a6442
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/7d0a6442
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/7d0a6442
Branch: refs/heads/feature-module
Commit: 7d0a6442e4c343dcc161d4ea84f9b205956ca9e5
Parents: ecb19fa
Author: ishark <is...@datatorrent.com>
Authored: Wed Oct 14 17:33:22 2015 -0700
Committer: ishark <is...@datatorrent.com>
Committed: Fri Oct 16 18:06:00 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/webapp/asm/CompactUtil.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/7d0a6442/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java b/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
index b0611b3..0b55578 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
@@ -153,12 +153,16 @@ public class CompactUtil
if (annotation.desc.contains("InputPortFieldAnnotation")
|| annotation.desc.contains("OutputPortFieldAnnotation")) {
List<Object> annotationValues = annotation.values;
- int index = 0;
- while (index < annotationValues.size()) {
- annotationMap.put((String) annotationValues.get(index++), annotationValues.get(index++));
+ if (annotationValues != null) {
+ int index = 0;
+ while (index <= annotationValues.size() - 2) {
+ String key = (String)annotationValues.get(index++);
+ Object value = annotationValues.get(index++);
+ annotationMap.put(key, value);
+ }
+ node.setAnnotations(annotationMap);
+ annotations.add(node);
}
- node.setAnnotations(annotationMap);
- annotations.add(node);
}
}
cfn.setVisibleAnnotations(annotations);
[18/36] incubator-apex-core git commit: APEX-189 Add original
copyright owner to NOTICE, rename license header template for clarity.
Posted by vr...@apache.org.
APEX-189 Add original copyright owner to NOTICE, rename license header template for clarity.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/529597fc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/529597fc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/529597fc
Branch: refs/heads/feature-module
Commit: 529597fc02ead77900ab661738405ca9b36e5e9b
Parents: d7afe72
Author: Thomas Weise <th...@datatorrent.com>
Authored: Sun Oct 18 22:33:20 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sun Oct 18 22:45:48 2015 -0700
----------------------------------------------------------------------
NOTICE | 3 +++
header.txt | 16 ++++++++++++++++
license.txt | 16 ----------------
pom.xml | 2 +-
4 files changed, 20 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/529597fc/NOTICE
----------------------------------------------------------------------
diff --git a/NOTICE b/NOTICE
index d834235..eef5e86 100644
--- a/NOTICE
+++ b/NOTICE
@@ -4,3 +4,6 @@ Copyright (c) 2015 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
+The initial developer of the original code is
+DataTorrent, Inc. (http://www.datatorrent.com)
+Copyright (c) 2012 - 2015. All Rights Reserved.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/529597fc/header.txt
----------------------------------------------------------------------
diff --git a/header.txt b/header.txt
new file mode 100644
index 0000000..60b675e
--- /dev/null
+++ b/header.txt
@@ -0,0 +1,16 @@
+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.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/529597fc/license.txt
----------------------------------------------------------------------
diff --git a/license.txt b/license.txt
deleted file mode 100644
index 60b675e..0000000
--- a/license.txt
+++ /dev/null
@@ -1,16 +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.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/529597fc/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 9cd4e5d..528eab4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -99,7 +99,7 @@
<artifactId>license-maven-plugin</artifactId>
<version>2.11</version>
<configuration>
- <header>license.txt</header>
+ <header>header.txt</header>
<properties>
<owner>Apache Software Foundation</owner>
<email>dev@apex.incubator.apache.org</email>
[05/36] incubator-apex-core git commit: Merge branch 'APEX-162' of
https://github.com/vrozov/incubator-apex-core into devel-3
Posted by vr...@apache.org.
Merge branch 'APEX-162' of https://github.com/vrozov/incubator-apex-core into devel-3
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/14f5269e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/14f5269e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/14f5269e
Branch: refs/heads/feature-module
Commit: 14f5269ef6bcdb5a27db0c88dd8bbc05cc73254e
Parents: 8767377 3c35ccc
Author: Thomas Weise <th...@datatorrent.com>
Authored: Tue Oct 13 18:23:00 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Tue Oct 13 18:23:00 2015 -0700
----------------------------------------------------------------------
engine/pom.xml | 2 +-
.../com/datatorrent/stram/CheckpointTest.java | 55 ++++--------
.../stram/LogicalPlanModificationTest.java | 34 ++++----
.../datatorrent/stram/OutputUnifiedTest.java | 29 ++++---
.../stram/StramLocalClusterTest.java | 17 ++--
.../datatorrent/stram/StramMiniClusterTest.java | 6 +-
.../datatorrent/stram/StramRecoveryTest.java | 48 +++++------
.../com/datatorrent/stram/StreamCodecTest.java | 51 ++---------
.../stram/StreamingContainerManagerTest.java | 90 +++++++-------------
.../stram/engine/AutoMetricTest.java | 22 ++---
.../stram/engine/ProcessingModeTests.java | 17 ++--
.../stram/plan/StreamPersistanceTests.java | 27 ++----
.../stram/support/StramTestSupport.java | 64 ++++++++++++--
13 files changed, 212 insertions(+), 250 deletions(-)
----------------------------------------------------------------------
[08/36] incubator-apex-core git commit: APEX-197 #resolve #comment
making the types of files excluded consistent
Posted by vr...@apache.org.
APEX-197 #resolve #comment making the types of files excluded consistent
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/330a2841
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/330a2841
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/330a2841
Branch: refs/heads/feature-module
Commit: 330a2841c51736bf1ac7aded90c46e1947b4f56f
Parents: ecb19fa
Author: Chandni Singh <cs...@apache.org>
Authored: Wed Oct 14 16:48:11 2015 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Wed Oct 14 16:48:11 2015 -0700
----------------------------------------------------------------------
pom.xml | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/330a2841/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index e75d3c3..c45f4e2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -102,11 +102,10 @@
<exclude>DISCLAIMER</exclude>
<exclude>LICENSE</exclude>
<exclude>NOTICE</exclude>
- <exclude>license.txt</exclude>
<exclude>**/*.md</exclude>
<exclude>**/*.txt</exclude>
+ <exclude>**/*.importorder</exclude>
<exclude>**/archetype-resources/**</exclude>
- <exclude>src/test/resources/projects/basic/goal.txt</exclude>
</excludes>
<mapping>
<dtcli>SCRIPT_STYLE</dtcli>
@@ -180,11 +179,11 @@
<exclude>.idea/**</exclude>
<exclude>**/src/test/resources/**/MANIFEST.MF</exclude>
<exclude>**/src/test/resources/**/*.json</exclude>
- <exclude>**/src/test/resources/**/*.txt</exclude>
<exclude>**/resources/META-INF/services/**</exclude>
- <exclude>CHANGELOG.md</exclude>
<exclude>**/archetype-resources/**</exclude>
- <exclude>README.md</exclude>
+ <exclude>**/*.md</exclude>
+ <exclude>**/*.txt</exclude>
+ <exclude>**/*.importorder</exclude>
</excludes>
</configuration>
</plugin>
[26/36] incubator-apex-core git commit: APEX-211 #resolve #comment
placement of left brace in static block
Posted by vr...@apache.org.
APEX-211 #resolve #comment placement of left brace in static block
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/eb0a95fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/eb0a95fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/eb0a95fe
Branch: refs/heads/feature-module
Commit: eb0a95fed5243495bdbecf6ea159467d92d698aa
Parents: 496161a
Author: Chandni Singh <cs...@apache.org>
Authored: Wed Oct 21 17:15:39 2015 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Wed Oct 21 17:17:04 2015 -0700
----------------------------------------------------------------------
apex_checks.xml | 4 ++--
api/pom.xml | 2 +-
common/pom.xml | 2 +-
engine/pom.xml | 2 +-
4 files changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/eb0a95fe/apex_checks.xml
----------------------------------------------------------------------
diff --git a/apex_checks.xml b/apex_checks.xml
index 0cba087..bc17e64 100644
--- a/apex_checks.xml
+++ b/apex_checks.xml
@@ -54,13 +54,13 @@
<module name="LeftCurly">
<property name="option" value="eol"/>
<property name="tokens"
- value="LITERAL_CATCH, LITERAL_DO, LITERAL_ELSE, LITERAL_FINALLY, LITERAL_FOR, LITERAL_IF, LITERAL_SWITCH, LITERAL_SYNCHRONIZED, LITERAL_TRY, LITERAL_WHILE"/>
+ value="LITERAL_CATCH, LITERAL_DO, LITERAL_ELSE, LITERAL_FINALLY, LITERAL_FOR, LITERAL_IF, LITERAL_SWITCH, LITERAL_SYNCHRONIZED, LITERAL_TRY, LITERAL_WHILE, STATIC_INIT"/>
</module>
<module name="LeftCurly">
<property name="option" value="nl"/>
<property name="tokens"
- value="ANNOTATION_DEF, CLASS_DEF, CTOR_DEF, ENUM_DEF, ENUM_CONSTANT_DEF, INTERFACE_DEF, METHOD_DEF, STATIC_INIT "/>
+ value="ANNOTATION_DEF, CLASS_DEF, CTOR_DEF, ENUM_DEF, ENUM_CONSTANT_DEF, INTERFACE_DEF, METHOD_DEF"/>
</module>
<module name="RightCurly">
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/eb0a95fe/api/pom.xml
----------------------------------------------------------------------
diff --git a/api/pom.xml b/api/pom.xml
index bb851bd..9224e63 100644
--- a/api/pom.xml
+++ b/api/pom.xml
@@ -89,7 +89,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>50</maxAllowedViolations>
+ <maxAllowedViolations>49</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/eb0a95fe/common/pom.xml
----------------------------------------------------------------------
diff --git a/common/pom.xml b/common/pom.xml
index b6929ab..1f7080d 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -74,7 +74,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>87</maxAllowedViolations>
+ <maxAllowedViolations>80</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/eb0a95fe/engine/pom.xml
----------------------------------------------------------------------
diff --git a/engine/pom.xml b/engine/pom.xml
index f5d8275..8e75151 100644
--- a/engine/pom.xml
+++ b/engine/pom.xml
@@ -145,7 +145,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>2076</maxAllowedViolations>
+ <maxAllowedViolations>2063</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
[27/36] incubator-apex-core git commit: APEX-204 #resolve #comment
setting linebreaks to true for NoWhitespaceBefore->dot and continuation
indentation to be 2
Posted by vr...@apache.org.
APEX-204 #resolve #comment setting linebreaks to true for NoWhitespaceBefore->dot and continuation indentation to be 2
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/496161ac
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/496161ac
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/496161ac
Branch: refs/heads/feature-module
Commit: 496161ac2d936e2c67f1ec6317cc440bebaf9148
Parents: e3cecc2
Author: Chandni Singh <cs...@apache.org>
Authored: Tue Oct 20 15:25:51 2015 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Wed Oct 21 17:17:04 2015 -0700
----------------------------------------------------------------------
apex_checks.xml | 10 +++++++++-
engine/pom.xml | 2 +-
pom.xml | 2 +-
3 files changed, 11 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/496161ac/apex_checks.xml
----------------------------------------------------------------------
diff --git a/apex_checks.xml b/apex_checks.xml
index 3bdc27e..0cba087 100644
--- a/apex_checks.xml
+++ b/apex_checks.xml
@@ -80,7 +80,12 @@
</module>
<module name="NoWhitespaceBefore">
- <property name="tokens" value="SEMI, DOT, POST_DEC, POST_INC"/>
+ <property name="tokens" value="COMMA, SEMI, POST_DEC, POST_INC"/>
+ </module>
+
+ <module name="NoWhitespaceBefore">
+ <property name="tokens" value="DOT"/>
+ <property name="allowLineBreaks" value="true"/>
</module>
<module name="WhitespaceAround">
@@ -133,6 +138,9 @@
<property name="basicOffset" value="2"/>
<property name="braceAdjustment" value="0"/>
<property name="caseIndent" value="2"/>
+ <property name="throwsIndent" value="2"/>
+ <property name="arrayInitIndent" value="2"/>
+ <property name="lineWrappingIndentation" value="2"/>
</module>
<module name="ImportOrder">
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/496161ac/engine/pom.xml
----------------------------------------------------------------------
diff --git a/engine/pom.xml b/engine/pom.xml
index 0c99ce0..f5d8275 100644
--- a/engine/pom.xml
+++ b/engine/pom.xml
@@ -145,7 +145,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>2320</maxAllowedViolations>
+ <maxAllowedViolations>2076</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/496161ac/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 13d5f52..807bf7f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -310,7 +310,7 @@
<dependency>
<groupId>com.puppycrawl.tools</groupId>
<artifactId>checkstyle</artifactId>
- <version>6.9</version>
+ <version>6.11</version>
</dependency>
</dependencies>
<executions>
[30/36] incubator-apex-core git commit: Produce .zip and .tar.gz
source release packages.
Posted by vr...@apache.org.
Produce .zip and .tar.gz source release packages.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/3b3d068f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/3b3d068f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/3b3d068f
Branch: refs/heads/feature-module
Commit: 3b3d068ff86180744fd5c7adb3d456c3c8d77ca3
Parents: 2e8ffc6
Author: Thomas Weise <th...@datatorrent.com>
Authored: Wed Oct 21 21:32:59 2015 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Thu Oct 22 00:08:32 2015 -0700
----------------------------------------------------------------------
pom.xml | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/3b3d068f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 807bf7f..f797805 100644
--- a/pom.xml
+++ b/pom.xml
@@ -359,6 +359,7 @@
<distMgmtReleaseId>apache.staging.https</distMgmtReleaseId>
<distMgmtReleaseUrl>https://repository.apache.org/service/local/staging/deploy/maven2</distMgmtReleaseUrl>
<distMgmtDevUrl>file://${project.build.directory}/mvn-repo</distMgmtDevUrl>
+ <sourceReleaseAssemblyDescriptor>source-release-zip-tar</sourceReleaseAssemblyDescriptor>
<package.prefix>/opt/datatorrent</package.prefix>
<package.groupname>dtorrent</package.groupname>
</properties>