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/05/01 21:46:48 UTC
[GitHub] [beam] ibzib opened a new pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
ibzib opened a new pull request #11591:
URL: https://github.com/apache/beam/pull/11591
I had to resolve a minor merge conflict between 325e0f1 and 8de324f22ca04b3716abf58ba77c2a3c117263a2.
------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622949604
retest this please
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622977564
(have removed the commit again)
----------------------------------------------------------------
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] mxm commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418944010
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
This is a batch-only change. Streaming already flushes timers on close. I don't now whether dropping processing timers is correct. I think we should continue to fire them as well.
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622936380
(Pushed a commit to fix Spark)
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-623339278
Merging. We can still follow up with the discussion though.
----------------------------------------------------------------
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] mxm commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r419292635
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
The code is now shared.
----------------------------------------------------------------
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] mxm commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418944102
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
I agree that the code could be shared. I initially thought about that when I pushed the fix in the other PR but got sidetracked..
----------------------------------------------------------------
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] ibzib commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622589337
Run Java Spark PortableValidatesRunner Batch
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-623152820
I've aggregated all the commits to fix the problems around this issue.
----------------------------------------------------------------
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] mxm commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418944102
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
I agree that the code could be shared with Spark. I initially thought about that when I pushed the fix in the other PR but got sidetracked..
----------------------------------------------------------------
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] ibzib commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418761275
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
Good question. Spark basically does the same thing here (ideally this code should be shared..) @mxm
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622939956
Had to fix spotless.
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-623152928
Run Python Spark ValidatesRunner
----------------------------------------------------------------
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] mxm commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r419292635
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
The code is now shared.
----------------------------------------------------------------
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] ibzib commented on a change in pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418762563
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
As expected, Spark is failing `test_pardo_timers`
----------------------------------------------------------------
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 #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418756561
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
Doesn't it make sense to make this change in batch as well for spark/flink in FlinkExecutableStageFunction and SparkExecutableStageFunction?
Any watermark based timers should continue to be eligible and continue to fire while in batch while the processing time timers should be dropped.
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622939975
Whoops, just saw that you opened https://github.com/apache/beam/pull/11595
----------------------------------------------------------------
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] mxm commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
mxm commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622936299
Run Java Spark PortableValidatesRunner Batch
----------------------------------------------------------------
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] ibzib commented on pull request #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #11591:
URL: https://github.com/apache/beam/pull/11591#issuecomment-622589471
Run Python Spark ValidatesRunner
----------------------------------------------------------------
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 #11591: [BEAM-9801] [cherry-pick] Pass in fire timestamp to timer callback
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11591:
URL: https://github.com/apache/beam/pull/11591#discussion_r418756561
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
##########
@@ -247,25 +247,27 @@ public void reduce(Iterable<WindowedValue<InputT>> iterable, Collector<RawUnionV
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
// Now we fire the timers and process elements generated by timers (which may be timers itself)
- try (RemoteBundle bundle =
- stageBundleFactory.getBundle(
- receiverFactory, timerReceiverFactory, stateRequestHandler, progressHandler)) {
-
- PipelineTranslatorUtils.fireEligibleTimers(
- timerInternals,
- (KV<String, String> transformAndTimerId, Timer<?> timerValue) -> {
- FnDataReceiver<Timer> fnTimerReceiver =
- bundle.getTimerReceivers().get(transformAndTimerId);
- Preconditions.checkNotNull(
- fnTimerReceiver, "No FnDataReceiver found for %s", transformAndTimerId);
- try {
- fnTimerReceiver.accept(timerValue);
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(Locale.ENGLISH, "Failed to process timer: %s", timerValue));
- }
- },
- currentTimerKey);
+ while (timerInternals.hasPendingTimers()) {
Review comment:
Doesn't it make sense to make this change in batch as well for spark/flink in FlinkExecutableStageFunction and SparkExecutableStageFunction
----------------------------------------------------------------
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