You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/06/04 18:15:25 UTC
[GitHub] [beam] lukecwik opened a new pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window.
lukecwik opened a new pull request #11922:
URL: https://github.com/apache/beam/pull/11922
This fixes a bug where we would output within all the windows instead of just the current window.
This would not impact any SDF that used only a single window while processing.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #11922:
URL: https://github.com/apache/beam/pull/11922#issuecomment-639757431
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436113097
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -296,6 +299,12 @@ public void accept(WindowedValue input) throws Exception {
/** Only valid during {@code processElement...} methods, null otherwise. */
private WindowedValue<InputT> currentElement;
+ /**
+ * Only valid during {@link #processElementForSizedElementAndRestriction} and {@link
+ * #processElementForSizedElementAndRestriction}.
Review comment:
Fixed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436116972
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -577,26 +586,83 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
switch (pTransform.getSpec().getUrn()) {
case PTransformTranslation.SPLITTABLE_PROCESS_ELEMENTS_URN:
this.convertSplitResultToWindowedSplitResult =
- (splitResult, watermarkEstimatorState) ->
- WindowedSplitResult.forRoots(
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()),
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getResidual(), watermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()));
+ (splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
Review comment:
Ah I see. It's different from what I know about an `Iterator`. Thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436111296
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -577,26 +586,83 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
switch (pTransform.getSpec().getUrn()) {
case PTransformTranslation.SPLITTABLE_PROCESS_ELEMENTS_URN:
this.convertSplitResultToWindowedSplitResult =
- (splitResult, watermarkEstimatorState) ->
- WindowedSplitResult.forRoots(
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()),
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getResidual(), watermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()));
+ (splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
+ return WindowedSplitResult.forRoots(
+ primaryFullyProcessedWindows.isEmpty()
+ ? null
+ : WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(currentRestriction, currentWatermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ primaryFullyProcessedWindows,
+ currentElement.getPane()),
+ WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ currentWindow,
+ currentElement.getPane()),
+ WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(splitResult.getResidual(), watermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ currentWindow,
+ currentElement.getPane()),
+ residualUnprocessedWindows.isEmpty()
+ ? null
+ : WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(currentRestriction, currentWatermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ residualUnprocessedWindows,
+ currentElement.getPane()));
+ };
break;
case PTransformTranslation.SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN:
this.convertSplitResultToWindowedSplitResult =
(splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
Review comment:
ImmutableList.copyOf(iterator) drains all the remaining elements from iterator. It is effectively:
```
while (iterator.hasNext()) {
list.add(iterator.next());
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r435546832
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -1593,6 +1601,391 @@ public void testProcessElementForSizedElementAndRestriction() throws Exception {
assertEquals(stateData, fakeClient.getData());
}
+ @Test
+ public void testProcessElementForWindowedSizedElementAndRestriction() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("unused"));
+ PCollectionView<String> singletonSideInputView = valuePCollection.apply(View.asSingleton());
+ TestSplittableDoFn doFn = new TestSplittableDoFn(singletonSideInputView);
+
+ valuePCollection
+ .apply(Window.into(SlidingWindows.of(Duration.standardSeconds(1))))
+ .apply(TEST_TRANSFORM_ID, ParDo.of(doFn).withSideInputs(singletonSideInputView));
+
+ RunnerApi.Pipeline pProto =
+ ProtoOverrides.updateTransform(
+ PTransformTranslation.PAR_DO_TRANSFORM_URN,
+ PipelineTranslation.toProto(p, SdkComponents.create(p.getOptions()), true),
+ SplittableParDoExpander.createSizedReplacement());
+ String expandedTransformId =
+ Iterables.find(
+ pProto.getComponents().getTransformsMap().entrySet(),
+ entry ->
+ entry
+ .getValue()
+ .getSpec()
+ .getUrn()
+ .equals(
+ PTransformTranslation
+ .SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN)
+ && entry.getValue().getUniqueName().contains(TEST_TRANSFORM_ID))
+ .getKey();
+ RunnerApi.PTransform pTransform =
+ pProto.getComponents().getTransformsOrThrow(expandedTransformId);
+ String inputPCollectionId =
+ pTransform.getInputsOrThrow(ParDoTranslation.getMainInputName(pTransform));
+ RunnerApi.PCollection inputPCollection =
+ pProto.getComponents().getPcollectionsOrThrow(inputPCollectionId);
+ RehydratedComponents rehydratedComponents =
+ RehydratedComponents.forComponents(pProto.getComponents());
+ Coder<WindowedValue> inputCoder =
+ WindowedValue.getFullCoder(
+ CoderTranslation.fromProto(
+ pProto.getComponents().getCodersOrThrow(inputPCollection.getCoderId()),
+ rehydratedComponents,
+ TranslationContext.DEFAULT),
+ (Coder)
+ CoderTranslation.fromProto(
+ pProto
+ .getComponents()
+ .getCodersOrThrow(
+ pProto
+ .getComponents()
+ .getWindowingStrategiesOrThrow(
+ inputPCollection.getWindowingStrategyId())
+ .getWindowCoderId()),
+ rehydratedComponents,
+ TranslationContext.DEFAULT));
+ String outputPCollectionId = pTransform.getOutputsOrThrow("output");
+
+ ImmutableMap<StateKey, ByteString> stateData =
+ ImmutableMap.of(
+ multimapSideInputKey(singletonSideInputView.getTagInternal().getId(), ByteString.EMPTY),
+ encode("8"));
+
+ FakeBeamFnStateClient fakeClient = new FakeBeamFnStateClient(stateData);
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add);
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+ List<ProgressRequestCallback> progressRequestCallbacks = new ArrayList<>();
+ BundleSplitListener.InMemory splitListener = BundleSplitListener.InMemory.create();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ fakeClient,
+ null /* beamFnTimerClient */,
+ TEST_TRANSFORM_ID,
+ pTransform,
+ Suppliers.ofInstance("57L")::get,
+ pProto.getComponents().getPcollectionsMap(),
+ pProto.getComponents().getCodersMap(),
+ pProto.getComponents().getWindowingStrategiesMap(),
+ consumers,
+ startFunctionRegistry,
+ finishFunctionRegistry,
+ teardownFunctions::add,
+ progressRequestCallbacks::add,
+ splitListener,
+ null /* bundleFinalizer */);
+
+ Iterables.getOnlyElement(startFunctionRegistry.getFunctions()).run();
+ mainOutputValues.clear();
+
+ assertThat(consumers.keySet(), containsInAnyOrder(inputPCollectionId, outputPCollectionId));
+
+ FnDataReceiver<WindowedValue<?>> mainInput =
+ consumers.getMultiplexingConsumer(inputPCollectionId);
+ assertThat(mainInput, instanceOf(HandlesSplits.class));
+
+ BoundedWindow window1 = new IntervalWindow(new Instant(5), new Instant(10));
+ BoundedWindow window2 = new IntervalWindow(new Instant(6), new Instant(11));
+ {
+ // Check that before processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+ WindowedValue<?> firstValue =
+ valueInWindows(
+ KV.of(
+ KV.of("5", KV.of(new OffsetRange(5, 10), GlobalWindow.TIMESTAMP_MIN_VALUE)), 5.0),
+ window1,
+ window2);
+ mainInput.accept(firstValue);
+ // Check that after processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+
+ // Since the side input upperBound is 8 we will process 5, 6, and 7 then checkpoint.
+ // We expect that the watermark advances to MIN + 7 and that the primary represents [5, 8)
+ // with the original watermark while the residual represents [8, 10) with the new MIN + 7
+ // watermark.
+ assertEquals(2, splitListener.getPrimaryRoots().size());
+ assertEquals(2, splitListener.getResidualRoots().size());
+ for (int i = 0; i < splitListener.getPrimaryRoots().size(); ++i) {
+ BundleApplication primaryRoot = splitListener.getPrimaryRoots().get(i);
+ DelayedBundleApplication residualRoot = splitListener.getResidualRoots().get(i);
+ assertEquals(ParDoTranslation.getMainInputName(pTransform), primaryRoot.getInputId());
+ assertEquals(TEST_TRANSFORM_ID, primaryRoot.getTransformId());
+ assertEquals(
+ ParDoTranslation.getMainInputName(pTransform),
+ residualRoot.getApplication().getInputId());
+ assertEquals(TEST_TRANSFORM_ID, residualRoot.getApplication().getTransformId());
+ Instant expectedOutputWatermark = GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7);
+ assertEquals(
+ ImmutableMap.of(
+ "output",
+ org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Timestamp.newBuilder()
+ .setSeconds(expectedOutputWatermark.getMillis() / 1000)
+ .setNanos((int) (expectedOutputWatermark.getMillis() % 1000) * 1000000)
+ .build()),
+ residualRoot.getApplication().getOutputWatermarksMap());
+ assertEquals(
+ org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Duration.newBuilder()
+ .setSeconds(54)
+ .setNanos(321000000)
+ .build(),
+ residualRoot.getRequestedTimeDelay());
+ }
+ assertThat(
+ Collections2.transform(
+ splitListener.getPrimaryRoots(), (root) -> decode(inputCoder, root.getElement())),
+ contains(
+ WindowedValue.of(
+ KV.of(
+ KV.of("5", KV.of(new OffsetRange(5, 8), GlobalWindow.TIMESTAMP_MIN_VALUE)),
+ 3.0),
+ firstValue.getTimestamp(),
+ window1,
+ firstValue.getPane()),
+ WindowedValue.of(
+ KV.of(
+ KV.of("5", KV.of(new OffsetRange(5, 8), GlobalWindow.TIMESTAMP_MIN_VALUE)),
+ 3.0),
+ firstValue.getTimestamp(),
+ window2,
+ firstValue.getPane())));
+ assertThat(
+ Collections2.transform(
+ splitListener.getResidualRoots(),
+ (root) -> decode(inputCoder, root.getApplication().getElement())),
+ contains(
+ WindowedValue.of(
+ KV.of(
+ KV.of(
+ "5",
+ KV.of(new OffsetRange(8, 10), GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7))),
+ 2.0),
+ firstValue.getTimestamp(),
+ window1,
+ firstValue.getPane()),
+ WindowedValue.of(
+ KV.of(
+ KV.of(
+ "5",
+ KV.of(new OffsetRange(8, 10), GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7))),
+ 2.0),
+ firstValue.getTimestamp(),
+ window2,
+ firstValue.getPane())));
+ splitListener.clear();
+
+ // Check that before processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+ WindowedValue<?> secondValue =
+ valueInWindows(
+ KV.of(
+ KV.of("2", KV.of(new OffsetRange(0, 2), GlobalWindow.TIMESTAMP_MIN_VALUE)), 2.0),
+ window1,
+ window2);
+ mainInput.accept(secondValue);
+ // Check that after processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+
+ assertThat(
+ mainOutputValues,
+ contains(
+ WindowedValue.of(
+ "5:5", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(5), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "5:6", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(6), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "5:7", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "5:5", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(5), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "5:6", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(6), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "5:7", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "2:0", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(0), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "2:1", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(1), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "2:0", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(0), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "2:1", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(1), window2, firstValue.getPane())));
+ assertTrue(splitListener.getPrimaryRoots().isEmpty());
+ assertTrue(splitListener.getResidualRoots().isEmpty());
+ mainOutputValues.clear();
+ }
+
+ {
+ // Setup and launch the trySplit thread.
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ Future<HandlesSplits.SplitResult> trySplitFuture =
+ executorService.submit(
+ () -> {
+ try {
+ doFn.waitForSplitElementToBeProcessed();
+ // Currently processing "3" out of range [0, 5) elements.
+ assertEquals(0.6, ((HandlesSplits) mainInput).getProgress(), 0.01);
+
+ // Check that during progressing of an element we report progress
+ List<MonitoringInfo> mis =
+ Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos();
+ MonitoringInfo.Builder expectedCompleted = MonitoringInfo.newBuilder();
+ expectedCompleted.setUrn(MonitoringInfoConstants.Urns.WORK_COMPLETED);
+ expectedCompleted.setType(MonitoringInfoConstants.TypeUrns.PROGRESS_TYPE);
+ expectedCompleted.putLabels(
+ MonitoringInfoConstants.Labels.PTRANSFORM, TEST_TRANSFORM_ID);
+ expectedCompleted.setPayload(
+ ByteString.copyFrom(
+ CoderUtils.encodeToByteArray(
+ IterableCoder.of(DoubleCoder.of()), Collections.singletonList(3.0))));
+ MonitoringInfo.Builder expectedRemaining = MonitoringInfo.newBuilder();
+ expectedRemaining.setUrn(MonitoringInfoConstants.Urns.WORK_REMAINING);
+ expectedRemaining.setType(MonitoringInfoConstants.TypeUrns.PROGRESS_TYPE);
+ expectedRemaining.putLabels(
+ MonitoringInfoConstants.Labels.PTRANSFORM, TEST_TRANSFORM_ID);
+ expectedRemaining.setPayload(
+ ByteString.copyFrom(
+ CoderUtils.encodeToByteArray(
+ IterableCoder.of(DoubleCoder.of()), Collections.singletonList(2.0))));
+ assertThat(
+ mis,
+ containsInAnyOrder(expectedCompleted.build(), expectedRemaining.build()));
+
+ return ((HandlesSplits) mainInput).trySplit(0);
+ } finally {
+ doFn.releaseWaitingProcessElementThread();
+ }
+ });
+
+ // Check that before processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+ WindowedValue<?> splitValue =
+ valueInWindows(
+ KV.of(
+ KV.of("7", KV.of(new OffsetRange(0, 5), GlobalWindow.TIMESTAMP_MIN_VALUE)), 2.0),
+ window1,
+ window2);
+ mainInput.accept(splitValue);
+ HandlesSplits.SplitResult trySplitResult = trySplitFuture.get();
+
+ // Check that after processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+
+ // Since the SPLIT_ELEMENT is 3 we will process 0, 1, 2, 3 then be split on the first window.
Review comment:
If the window is observed we need to process the element for each window independently. You do point out an issue though that the split's primary and residual roots need to account for what windows were processed and which are still remaining.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436112241
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -941,20 +1047,59 @@ private Progress getProgress() {
convertSplitResultToWindowedSplitResult.apply(result, watermarkAndState.getValue());
}
+ List<BundleApplication> primaryRoots = new ArrayList<>();
Review comment:
Yes. As shown in the test the self checkpoint will checkpoint for the "remaining" windows as well.
I would like to leave it as is until we can build consensus around what we should do in these use cases.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436111186
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -577,26 +586,83 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
switch (pTransform.getSpec().getUrn()) {
case PTransformTranslation.SPLITTABLE_PROCESS_ELEMENTS_URN:
this.convertSplitResultToWindowedSplitResult =
- (splitResult, watermarkEstimatorState) ->
- WindowedSplitResult.forRoots(
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()),
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getResidual(), watermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()));
+ (splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
Review comment:
ImmutableList.copyOf(iterator) drains all the remaining elements from iterator. It is effectively:
```
while (iterator.hasNext()) {
list.add(iterator.next());
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #11922:
URL: https://github.com/apache/beam/pull/11922#issuecomment-639020928
R: @boyuanzz @chamikaramj
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436072548
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -296,6 +299,12 @@ public void accept(WindowedValue input) throws Exception {
/** Only valid during {@code processElement...} methods, null otherwise. */
private WindowedValue<InputT> currentElement;
+ /**
+ * Only valid during {@link #processElementForSizedElementAndRestriction} and {@link
+ * #processElementForSizedElementAndRestriction}.
Review comment:
Duplicated {@link #processElementForSizedElementAndRestriction}?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -577,26 +586,83 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
switch (pTransform.getSpec().getUrn()) {
case PTransformTranslation.SPLITTABLE_PROCESS_ELEMENTS_URN:
this.convertSplitResultToWindowedSplitResult =
- (splitResult, watermarkEstimatorState) ->
- WindowedSplitResult.forRoots(
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()),
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getResidual(), watermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()));
+ (splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
Review comment:
`currentWindowIterator .next()`?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -941,20 +1047,59 @@ private Progress getProgress() {
convertSplitResultToWindowedSplitResult.apply(result, watermarkAndState.getValue());
}
+ List<BundleApplication> primaryRoots = new ArrayList<>();
Review comment:
This also takes care of self checkpoint, right?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -577,26 +586,83 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
switch (pTransform.getSpec().getUrn()) {
case PTransformTranslation.SPLITTABLE_PROCESS_ELEMENTS_URN:
this.convertSplitResultToWindowedSplitResult =
- (splitResult, watermarkEstimatorState) ->
- WindowedSplitResult.forRoots(
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()),
- WindowedValue.of(
- KV.of(
- currentElement.getValue(),
- KV.of(splitResult.getResidual(), watermarkEstimatorState)),
- currentElement.getTimestamp(),
- currentWindow,
- currentElement.getPane()));
+ (splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
+ return WindowedSplitResult.forRoots(
+ primaryFullyProcessedWindows.isEmpty()
+ ? null
+ : WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(currentRestriction, currentWatermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ primaryFullyProcessedWindows,
+ currentElement.getPane()),
+ WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(splitResult.getPrimary(), currentWatermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ currentWindow,
+ currentElement.getPane()),
+ WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(splitResult.getResidual(), watermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ currentWindow,
+ currentElement.getPane()),
+ residualUnprocessedWindows.isEmpty()
+ ? null
+ : WindowedValue.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(currentRestriction, currentWatermarkEstimatorState)),
+ currentElement.getTimestamp(),
+ residualUnprocessedWindows,
+ currentElement.getPane()));
+ };
break;
case PTransformTranslation.SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN:
this.convertSplitResultToWindowedSplitResult =
(splitResult, watermarkEstimatorState) -> {
+ List<BoundedWindow> primaryFullyProcessedWindows =
+ ImmutableList.copyOf(
+ Iterables.limit(
+ currentElement.getWindows(), currentWindowIterator.previousIndex()));
+ // Advances the iterator consuming the remaining windows.
+ List<BoundedWindow> residualUnprocessedWindows =
+ ImmutableList.copyOf(currentWindowIterator);
Review comment:
`currentWindowIterator.next()`?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -632,6 +698,17 @@ public Object restriction() {
}
});
return WindowedSplitResult.forRoots(
+ primaryFullyProcessedWindows.isEmpty()
+ ? null
+ : WindowedValue.of(
+ KV.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(currentRestriction, currentWatermarkEstimatorState)),
+ fullSize),
Review comment:
I'm wondering should the size be `fullSize` * num of windows?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r436111838
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -632,6 +698,17 @@ public Object restriction() {
}
});
return WindowedSplitResult.forRoots(
+ primaryFullyProcessedWindows.isEmpty()
+ ? null
+ : WindowedValue.of(
+ KV.of(
+ KV.of(
+ currentElement.getValue(),
+ KV.of(currentRestriction, currentWatermarkEstimatorState)),
+ fullSize),
Review comment:
I don't think so but this warrants a larger discussion about what does an element+restriction in multiple windows mean and how that it impacts splitting/sizing.
I would like to leave it as is until we can build that consensus.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik merged pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
lukecwik merged pull request #11922:
URL: https://github.com/apache/beam/pull/11922
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11922: [BEAM-2939] Fix FnApiDoFnRunner to ensure that we output within the correct window when processing a splittable dofn
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11922:
URL: https://github.com/apache/beam/pull/11922#discussion_r435521657
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -1593,6 +1601,391 @@ public void testProcessElementForSizedElementAndRestriction() throws Exception {
assertEquals(stateData, fakeClient.getData());
}
+ @Test
+ public void testProcessElementForWindowedSizedElementAndRestriction() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("unused"));
+ PCollectionView<String> singletonSideInputView = valuePCollection.apply(View.asSingleton());
+ TestSplittableDoFn doFn = new TestSplittableDoFn(singletonSideInputView);
+
+ valuePCollection
+ .apply(Window.into(SlidingWindows.of(Duration.standardSeconds(1))))
+ .apply(TEST_TRANSFORM_ID, ParDo.of(doFn).withSideInputs(singletonSideInputView));
+
+ RunnerApi.Pipeline pProto =
+ ProtoOverrides.updateTransform(
+ PTransformTranslation.PAR_DO_TRANSFORM_URN,
+ PipelineTranslation.toProto(p, SdkComponents.create(p.getOptions()), true),
+ SplittableParDoExpander.createSizedReplacement());
+ String expandedTransformId =
+ Iterables.find(
+ pProto.getComponents().getTransformsMap().entrySet(),
+ entry ->
+ entry
+ .getValue()
+ .getSpec()
+ .getUrn()
+ .equals(
+ PTransformTranslation
+ .SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN)
+ && entry.getValue().getUniqueName().contains(TEST_TRANSFORM_ID))
+ .getKey();
+ RunnerApi.PTransform pTransform =
+ pProto.getComponents().getTransformsOrThrow(expandedTransformId);
+ String inputPCollectionId =
+ pTransform.getInputsOrThrow(ParDoTranslation.getMainInputName(pTransform));
+ RunnerApi.PCollection inputPCollection =
+ pProto.getComponents().getPcollectionsOrThrow(inputPCollectionId);
+ RehydratedComponents rehydratedComponents =
+ RehydratedComponents.forComponents(pProto.getComponents());
+ Coder<WindowedValue> inputCoder =
+ WindowedValue.getFullCoder(
+ CoderTranslation.fromProto(
+ pProto.getComponents().getCodersOrThrow(inputPCollection.getCoderId()),
+ rehydratedComponents,
+ TranslationContext.DEFAULT),
+ (Coder)
+ CoderTranslation.fromProto(
+ pProto
+ .getComponents()
+ .getCodersOrThrow(
+ pProto
+ .getComponents()
+ .getWindowingStrategiesOrThrow(
+ inputPCollection.getWindowingStrategyId())
+ .getWindowCoderId()),
+ rehydratedComponents,
+ TranslationContext.DEFAULT));
+ String outputPCollectionId = pTransform.getOutputsOrThrow("output");
+
+ ImmutableMap<StateKey, ByteString> stateData =
+ ImmutableMap.of(
+ multimapSideInputKey(singletonSideInputView.getTagInternal().getId(), ByteString.EMPTY),
+ encode("8"));
+
+ FakeBeamFnStateClient fakeClient = new FakeBeamFnStateClient(stateData);
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add);
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+ List<ProgressRequestCallback> progressRequestCallbacks = new ArrayList<>();
+ BundleSplitListener.InMemory splitListener = BundleSplitListener.InMemory.create();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ fakeClient,
+ null /* beamFnTimerClient */,
+ TEST_TRANSFORM_ID,
+ pTransform,
+ Suppliers.ofInstance("57L")::get,
+ pProto.getComponents().getPcollectionsMap(),
+ pProto.getComponents().getCodersMap(),
+ pProto.getComponents().getWindowingStrategiesMap(),
+ consumers,
+ startFunctionRegistry,
+ finishFunctionRegistry,
+ teardownFunctions::add,
+ progressRequestCallbacks::add,
+ splitListener,
+ null /* bundleFinalizer */);
+
+ Iterables.getOnlyElement(startFunctionRegistry.getFunctions()).run();
+ mainOutputValues.clear();
+
+ assertThat(consumers.keySet(), containsInAnyOrder(inputPCollectionId, outputPCollectionId));
+
+ FnDataReceiver<WindowedValue<?>> mainInput =
+ consumers.getMultiplexingConsumer(inputPCollectionId);
+ assertThat(mainInput, instanceOf(HandlesSplits.class));
+
+ BoundedWindow window1 = new IntervalWindow(new Instant(5), new Instant(10));
+ BoundedWindow window2 = new IntervalWindow(new Instant(6), new Instant(11));
+ {
+ // Check that before processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+ WindowedValue<?> firstValue =
+ valueInWindows(
+ KV.of(
+ KV.of("5", KV.of(new OffsetRange(5, 10), GlobalWindow.TIMESTAMP_MIN_VALUE)), 5.0),
+ window1,
+ window2);
+ mainInput.accept(firstValue);
+ // Check that after processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+
+ // Since the side input upperBound is 8 we will process 5, 6, and 7 then checkpoint.
+ // We expect that the watermark advances to MIN + 7 and that the primary represents [5, 8)
+ // with the original watermark while the residual represents [8, 10) with the new MIN + 7
+ // watermark.
+ assertEquals(2, splitListener.getPrimaryRoots().size());
+ assertEquals(2, splitListener.getResidualRoots().size());
+ for (int i = 0; i < splitListener.getPrimaryRoots().size(); ++i) {
+ BundleApplication primaryRoot = splitListener.getPrimaryRoots().get(i);
+ DelayedBundleApplication residualRoot = splitListener.getResidualRoots().get(i);
+ assertEquals(ParDoTranslation.getMainInputName(pTransform), primaryRoot.getInputId());
+ assertEquals(TEST_TRANSFORM_ID, primaryRoot.getTransformId());
+ assertEquals(
+ ParDoTranslation.getMainInputName(pTransform),
+ residualRoot.getApplication().getInputId());
+ assertEquals(TEST_TRANSFORM_ID, residualRoot.getApplication().getTransformId());
+ Instant expectedOutputWatermark = GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7);
+ assertEquals(
+ ImmutableMap.of(
+ "output",
+ org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Timestamp.newBuilder()
+ .setSeconds(expectedOutputWatermark.getMillis() / 1000)
+ .setNanos((int) (expectedOutputWatermark.getMillis() % 1000) * 1000000)
+ .build()),
+ residualRoot.getApplication().getOutputWatermarksMap());
+ assertEquals(
+ org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Duration.newBuilder()
+ .setSeconds(54)
+ .setNanos(321000000)
+ .build(),
+ residualRoot.getRequestedTimeDelay());
+ }
+ assertThat(
+ Collections2.transform(
+ splitListener.getPrimaryRoots(), (root) -> decode(inputCoder, root.getElement())),
+ contains(
+ WindowedValue.of(
+ KV.of(
+ KV.of("5", KV.of(new OffsetRange(5, 8), GlobalWindow.TIMESTAMP_MIN_VALUE)),
+ 3.0),
+ firstValue.getTimestamp(),
+ window1,
+ firstValue.getPane()),
+ WindowedValue.of(
+ KV.of(
+ KV.of("5", KV.of(new OffsetRange(5, 8), GlobalWindow.TIMESTAMP_MIN_VALUE)),
+ 3.0),
+ firstValue.getTimestamp(),
+ window2,
+ firstValue.getPane())));
+ assertThat(
+ Collections2.transform(
+ splitListener.getResidualRoots(),
+ (root) -> decode(inputCoder, root.getApplication().getElement())),
+ contains(
+ WindowedValue.of(
+ KV.of(
+ KV.of(
+ "5",
+ KV.of(new OffsetRange(8, 10), GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7))),
+ 2.0),
+ firstValue.getTimestamp(),
+ window1,
+ firstValue.getPane()),
+ WindowedValue.of(
+ KV.of(
+ KV.of(
+ "5",
+ KV.of(new OffsetRange(8, 10), GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7))),
+ 2.0),
+ firstValue.getTimestamp(),
+ window2,
+ firstValue.getPane())));
+ splitListener.clear();
+
+ // Check that before processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+ WindowedValue<?> secondValue =
+ valueInWindows(
+ KV.of(
+ KV.of("2", KV.of(new OffsetRange(0, 2), GlobalWindow.TIMESTAMP_MIN_VALUE)), 2.0),
+ window1,
+ window2);
+ mainInput.accept(secondValue);
+ // Check that after processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+
+ assertThat(
+ mainOutputValues,
+ contains(
+ WindowedValue.of(
+ "5:5", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(5), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "5:6", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(6), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "5:7", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "5:5", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(5), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "5:6", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(6), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "5:7", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(7), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "2:0", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(0), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "2:1", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(1), window1, firstValue.getPane()),
+ WindowedValue.of(
+ "2:0", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(0), window2, firstValue.getPane()),
+ WindowedValue.of(
+ "2:1", GlobalWindow.TIMESTAMP_MIN_VALUE.plus(1), window2, firstValue.getPane())));
+ assertTrue(splitListener.getPrimaryRoots().isEmpty());
+ assertTrue(splitListener.getResidualRoots().isEmpty());
+ mainOutputValues.clear();
+ }
+
+ {
+ // Setup and launch the trySplit thread.
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ Future<HandlesSplits.SplitResult> trySplitFuture =
+ executorService.submit(
+ () -> {
+ try {
+ doFn.waitForSplitElementToBeProcessed();
+ // Currently processing "3" out of range [0, 5) elements.
+ assertEquals(0.6, ((HandlesSplits) mainInput).getProgress(), 0.01);
+
+ // Check that during progressing of an element we report progress
+ List<MonitoringInfo> mis =
+ Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos();
+ MonitoringInfo.Builder expectedCompleted = MonitoringInfo.newBuilder();
+ expectedCompleted.setUrn(MonitoringInfoConstants.Urns.WORK_COMPLETED);
+ expectedCompleted.setType(MonitoringInfoConstants.TypeUrns.PROGRESS_TYPE);
+ expectedCompleted.putLabels(
+ MonitoringInfoConstants.Labels.PTRANSFORM, TEST_TRANSFORM_ID);
+ expectedCompleted.setPayload(
+ ByteString.copyFrom(
+ CoderUtils.encodeToByteArray(
+ IterableCoder.of(DoubleCoder.of()), Collections.singletonList(3.0))));
+ MonitoringInfo.Builder expectedRemaining = MonitoringInfo.newBuilder();
+ expectedRemaining.setUrn(MonitoringInfoConstants.Urns.WORK_REMAINING);
+ expectedRemaining.setType(MonitoringInfoConstants.TypeUrns.PROGRESS_TYPE);
+ expectedRemaining.putLabels(
+ MonitoringInfoConstants.Labels.PTRANSFORM, TEST_TRANSFORM_ID);
+ expectedRemaining.setPayload(
+ ByteString.copyFrom(
+ CoderUtils.encodeToByteArray(
+ IterableCoder.of(DoubleCoder.of()), Collections.singletonList(2.0))));
+ assertThat(
+ mis,
+ containsInAnyOrder(expectedCompleted.build(), expectedRemaining.build()));
+
+ return ((HandlesSplits) mainInput).trySplit(0);
+ } finally {
+ doFn.releaseWaitingProcessElementThread();
+ }
+ });
+
+ // Check that before processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+ WindowedValue<?> splitValue =
+ valueInWindows(
+ KV.of(
+ KV.of("7", KV.of(new OffsetRange(0, 5), GlobalWindow.TIMESTAMP_MIN_VALUE)), 2.0),
+ window1,
+ window2);
+ mainInput.accept(splitValue);
+ HandlesSplits.SplitResult trySplitResult = trySplitFuture.get();
+
+ // Check that after processing an element we don't report progress
+ assertThat(Iterables.getOnlyElement(progressRequestCallbacks).getMonitoringInfos(), empty());
+
+ // Since the SPLIT_ELEMENT is 3 we will process 0, 1, 2, 3 then be split on the first window.
Review comment:
I'm wondering why the split is on the first window?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org