Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/163#discussion_r45312032
--- Diff:
engine/src/test/java/com/datatorrent/stram/engine/GenericNodeTest.java ---
@@ -296,4 +345,132 @@ public void run()
Assert.assertTrue("End window not called", go.endWindowId !=
go.beginWindowId);
}
+ /**
+ * Tests to make sure that {@link CheckpointListener#checkpointed(long)
is called
+ * immediately after copying to hdfs in the case of exactly once
processing mode.
+ * @throws Exception
+ */
+ @Test
+ public void testCheckpointedCallTestExactlyOnce() throws Exception
+ {
+ asyncCheckpointedCalledHelper(ProcessingMode.EXACTLY_ONCE);
+ }
+
+ /**
+ * Tests to make sure that {@link CheckpointListener#checkpointed(long)
is NOT called immediately after the asynchronous
+ * task of copying to hdfs is triggered, in the at least once processing
mode.
+ * @throws Exception
+ */
+ @Test
+ public void testCheckpointedCallTestAtleastOnce() throws Exception
+ {
+ asyncCheckpointedCalledHelper(ProcessingMode.AT_LEAST_ONCE);
+ }
+
+ /**
+ * Tests to make sure that {@link CheckpointListener#checkpointed(long)
is NOT called immediately after the asynchronous
+ * task of copying to hdfs is triggered, in the at most once processing
mode.
+ * @throws Exception
+ */
+ @Test
+ public void testCheckpointedCallTestAtModeOnce() throws Exception
+ {
+ asyncCheckpointedCalledHelper(ProcessingMode.AT_MOST_ONCE);
+ }
+
+ private void asyncCheckpointedCalledHelper(ProcessingMode
processingMode) throws Exception
+ {
+ long maxSleep = 5000;
+ long sleeptime = 25L;
+ GenericOperator go = new GenericOperator();
+ AsyncFSStorageAgent storageAgent = new
TestAsyncFSStorageAgent(testMeta.getDir(), new Configuration());
+ DefaultAttributeMap amap = new DefaultAttributeMap();
+ amap.put(OperatorContext.STORAGE_AGENT, storageAgent);
+ amap.put(OperatorContext.PROCESSING_MODE, processingMode);
+ final GenericNode gn = new GenericNode(go, new
com.datatorrent.stram.engine.OperatorContext(0, amap, null));
+ gn.setId(1);
+ gn.APPLICATION_WINDOW_COUNT = 1;
+ gn.CHECKPOINT_WINDOW_COUNT = 1;
+ DefaultReservoir reservoir1 = new DefaultReservoir("ip1Res", 1024);
+ DefaultReservoir reservoir2 = new DefaultReservoir("ip2Res", 1024);
+
+ gn.connectInputPort("ip1", reservoir1);
+ gn.connectInputPort("ip2", reservoir2);
+ gn.connectOutputPort("op", Sink.BLACKHOLE);
+
+ final AtomicBoolean ab = new AtomicBoolean(false);
+ Thread t = new Thread()
+ {
+ @Override
+ public void run()
+ {
+ ab.set(true);
+ gn.activate();
+ gn.run();
+ gn.deactivate();
+ }
+
+ };
+ t.start();
+
+ long interval = 0;
+ do {
+ Thread.sleep(sleeptime);
+ interval += sleeptime;
+ }
+ while ((ab.get() == false) && (interval < maxSleep));
+
+
+ int controlTupleCount = gn.controlTupleCount;
+ Tuple beginWindow1 = new Tuple(MessageType.BEGIN_WINDOW, 0x1L);
+
+ reservoir1.add(beginWindow1);
+ reservoir2.add(beginWindow1);
+
+ interval = 0;
+ do {
+ Thread.sleep(sleeptime);
+ interval += sleeptime;
+ }
+ while ((gn.controlTupleCount == controlTupleCount) && (interval <
maxSleep));
+ Assert.assertTrue("Begin window called", go.endWindowId !=
go.beginWindowId);
+ controlTupleCount = gn.controlTupleCount;
+
+ Tuple endWindow1 = new EndWindowTuple(0x1L);
+
+ gn.doCheckpoint = true;
+
+ reservoir1.add(endWindow1);
+ reservoir2.add(endWindow1);
+
+ interval = 0;
+ do {
+ Thread.sleep(sleeptime);
+ interval += sleeptime;
+ }
+ while ((gn.controlTupleCount == controlTupleCount) && (interval <
maxSleep));
+ Assert.assertTrue("End window called", go.endWindowId ==
go.beginWindowId);
+
+ if (processingMode == ProcessingMode.EXACTLY_ONCE) {
+ Assert.assertTrue(go.checkpointed);
+ } else {
+ Assert.assertFalse(go.checkpointed);
+ }
+ }
--- End diff --
There is no endStreamTuple passed or shutdown call on GenericNode gn
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
