You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by sandeshh <gi...@git.apache.org> on 2015/12/16 00:52:13 UTC
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
GitHub user sandeshh opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/136
*For Review Only* Apex 247 - Implementing Delay operator
@davidyan74 Please review. I
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/sandeshh/incubator-apex-malhar APEX-247
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/136.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #136
----
commit e58e246e5f0a20a8feec0febce4123f0be729d86
Author: David Yan <da...@datatorrent.com>
Date: 2015-11-19T21:20:38Z
APEX-60 added iteration demo
commit ecfd683ef452a1a846ac73880ae6135bbe793bec
Author: David Yan <da...@datatorrent.com>
Date: 2015-12-08T17:11:56Z
simulate write ahead log
commit 1f46b34c551800a6ee564b0bbabaa9fcb3ea8431
Author: sandeshh <sa...@gmail.com>
Date: 2015-12-15T23:47:52Z
Apex 247- Delay operator with recovery support
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48882086
--- Diff: demos/iteration/src/test/resources/log4j.properties ---
@@ -0,0 +1,40 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+log4j.rootLogger=DEBUG,CONSOLE
+
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} %M - %m%n
+
+log4j.appender.RFA=org.apache.log4j.RollingFileAppender
+log4j.appender.RFA.layout=org.apache.log4j.PatternLayout
+log4j.appender.RFA.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} %M - %m%n
+log4j.appender.RFA.File=/tmp/app.log
+
+# to enable, add SYSLOG to rootLogger
+log4j.appender.SYSLOG=org.apache.log4j.net.SyslogAppender
+log4j.appender.SYSLOG.syslogHost=127.0.0.1
+log4j.appender.SYSLOG.layout=org.apache.log4j.PatternLayout
+log4j.appender.SYSLOG.layout.conversionPattern=${dt.cid} %-5p [%t] %c{2} %x - %m%n
+log4j.appender.SYSLOG.Facility=LOCAL1
--- End diff --
Do you need RFA and SYSLOG?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48248563
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DefaultDelayOperator.java ---
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
--- End diff --
Java Docs?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r47742770
--- Diff: library/src/test/java/com/datatorrent/lib/iteration/DelayOperatorTest.java ---
@@ -0,0 +1,174 @@
+package com.datatorrent.lib.iteration;
--- End diff --
license header?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r47740376
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DelayOperator.java ---
@@ -0,0 +1,130 @@
+package com.datatorrent.lib.iteration;
--- End diff --
licence header
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48248566
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DefaultDelayOperator.java ---
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+public class DefaultDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
--- End diff --
Provide java docs for getter and setter methods.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48248583
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DefaultDelayOperator.java ---
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+public class DefaultDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private WindowDataManager windowDataManager;
+ private transient long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient boolean timeToStoreTheWindow = false;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ DefaultDelayOperator()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow(long windowId)
+ {
+ replay(windowId);
+ }
+
+ private void replay( long windowId )
+ {
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ processTuple(tuple);
+ }
+ } catch (IOException e) {
--- End diff --
Why re-throwing the exception?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by sandeshh <gi...@git.apache.org>.
Github user sandeshh closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/136
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48248570
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DefaultDelayOperator.java ---
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+public class DefaultDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private WindowDataManager windowDataManager;
+ private transient long currentWindowId;
--- End diff --
Can you make the variables to protected, which of the variables doesn't have setter and getter ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48248576
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DefaultDelayOperator.java ---
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+public class DefaultDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private WindowDataManager windowDataManager;
+ private transient long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient boolean timeToStoreTheWindow = false;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ DefaultDelayOperator()
+ {
--- End diff --
Assign default value to windowDataManager.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r48248573
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DefaultDelayOperator.java ---
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+public class DefaultDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private WindowDataManager windowDataManager;
+ private transient long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient boolean timeToStoreTheWindow = false;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
--- End diff --
Access specifier is missed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *For Review Only* Apex 247 - I...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/136#discussion_r47740384
--- Diff: demos/iteration/src/test/java/com/datatorrent/demos/iteration/ApplicationTest.java ---
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.datatorrent.demos.iteration;
+
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+import com.datatorrent.api.LocalMode;
+
+
+/**
+ *
+ */
+public class ApplicationTest
+{
+ @Test
+ public void testSomeMethod() throws Exception
+ {
+ LocalMode lma = LocalMode.newInstance();
--- End diff --
Indentation is off
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---