johnyangk commented on a change in pull request #137: [NEMO-232] Implement
InputWatermarkManager
URL: https://github.com/apache/incubator-nemo/pull/137#discussion_r229599734
##########
File path:
runtime/executor/src/test/java/org/apache/nemo/runtime/executor/task/TaskExecutorTest.java
##########
@@ -601,6 +600,91 @@ public void close() {
}
}
+ /**
+ * Source vertex for unbounded source test.
+ */
+ private final class TestUnboundedSourceVertex extends SourceVertex {
+
+ @Override
+ public boolean isBounded() {
+ return false;
+ }
+
+ @Override
+ public List<Readable> getReadables(int desiredNumOfSplits) throws
Exception {
+ return null;
+ }
+
+ @Override
+ public void clearInternalStates() {
+
+ }
+
+ @Override
+ public IRVertex getClone() {
+ return null;
+ }
+ }
+
+ private final class TestUnboundedSourceReadable implements Readable {
+ int pointer = 0;
+ final int middle = elements.size() / 2;
+ final int end = elements.size();
+ boolean watermarkEmitted = false;
+ final List<Long> watermarks;
+ int numEmittedWatermarks = 0;
+
+ public TestUnboundedSourceReadable(final List<Long> watermarks) {
+ this.watermarks = watermarks;
+ }
+
+ @Override
+ public void prepare() {
+
+ }
+
+ // This emulates unbounded source that throws NoSuchElementException
+ // It reads current data until middle point and throws
NoSuchElementException at the middle point.
+ // It resumes the data reading after emitting a watermark, and finishes at
the end of the data.
+ @Override
+ public Object readCurrent() throws NoSuchElementException {
+ if (pointer == middle && !watermarkEmitted) {
+ throw new NoSuchElementException();
+ }
+
+ return elements.get(pointer);
+ }
+
+ @Override
+ public void advance() throws IOException {
+ pointer += 1;
+ }
+
+ @Override
+ public long readWatermark() {
+ if (numEmittedWatermarks == watermarks.size() - 1) {
+ watermarkEmitted = true;
Review comment:
watermarkEmitted => allWatermarkEmitted?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
