[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[
https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16647630#comment-16647630
]
Maximilian Michels commented on BEAM-5713:
--
{quote}
I added more tasks and they are all squeezed into the same slots (only 8 out of
144 task slots are used).
{quote}
Tasks which directly depend on each other share the same task slot. That is how
pipelining in Flink works. AFAIK pipelines can get arbitrarily long.
{quote}
The scheduling of all tasks to the same slot is consistent, distribution over
hosts isn't. With parallelism 4, different result (multiple hosts).
{quote}
Is that consistent behavior for parallelism of 4? I find that it depends on
what the iterator returns from the task slot HashMap. This depends on a number
of factors, e.g. what jobs you ran before, how TaskManager registered.
> Flink portable runner schedules all tasks of streaming job on same task
> manager
> ---
>
> Key: BEAM-5713
> URL: https://issues.apache.org/jira/browse/BEAM-5713
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Affects Versions: 2.8.0
>Reporter: Thomas Weise
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Attachments: Different SlotSharingGroup.png, With
> RichParallelSourceFunction and parallelism 5.png,
> image-2018-10-11-11-43-50-333.png, image-2018-10-11-16-20-45-221.png
>
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> The cluster has 9 task managers and 144 task slots total. A simple streaming
> pipeline with parallelism of 8 will get all tasks scheduled on the same task
> manager, causing the host to be fully booked and the remaining cluster idle.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646815#comment-16646815 ] Maximilian Michels commented on BEAM-5713: -- After testing this with native Flink jobs, looking inside the Scheduler code, and checking out FLINK-1003, it is clear there is no round-robin task scheduling logic for distributing tasks across TaskManagers. The location of task slots is transparent for normal operators. If you have more task slots per TaskManagers than tasks, then very likely all tasks will end up on the same TaskManager. If the cluster is sized to the Job, this shouldn't impact performance. Also, if all task slots are filled by multiple jobs, this should be fine. It is only problematic if the cluster is not fully utilized. Then, spreading the load across nodes should lead to a better performance. > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Attachments: Different SlotSharingGroup.png, With > RichParallelSourceFunction and parallelism 5.png > > Time Spent: 0.5h > Remaining Estimate: 0h > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5713: - Attachment: Different SlotSharingGroup.png > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Attachments: Different SlotSharingGroup.png, With > RichParallelSourceFunction and parallelism 5.png > > Time Spent: 20m > Remaining Estimate: 0h > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[
https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646497#comment-16646497
]
Maximilian Michels commented on BEAM-5713:
--
Putting the native source into a different {{SlotSharingGroup}} yields the
following:
!Different SlotSharingGroup.png!
> Flink portable runner schedules all tasks of streaming job on same task
> manager
> ---
>
> Key: BEAM-5713
> URL: https://issues.apache.org/jira/browse/BEAM-5713
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Affects Versions: 2.8.0
>Reporter: Thomas Weise
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Attachments: Different SlotSharingGroup.png, With
> RichParallelSourceFunction and parallelism 5.png
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> The cluster has 9 task managers and 144 task slots total. A simple streaming
> pipeline with parallelism of 8 will get all tasks scheduled on the same task
> manager, causing the host to be fully booked and the remaining cluster idle.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646479#comment-16646479 ] Maximilian Michels commented on BEAM-5713: -- Seems like this was largely by chance because your example doesn't use Impulse transforms. However, I've opened a PR which fixes sources written via portable DoFns. > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Attachments: With RichParallelSourceFunction and parallelism 5.png > > Time Spent: 20m > Remaining Estimate: 0h > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646345#comment-16646345 ] Maximilian Michels commented on BEAM-5713: -- !With RichParallelSourceFunction and parallelism 5.png! The load is not distributed completely equal across two TaskManagers, but it is an improvement. > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Attachments: With RichParallelSourceFunction and parallelism 5.png > > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5713: - Attachment: With RichParallelSourceFunction and parallelism 5.png > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Attachments: With RichParallelSourceFunction and parallelism 5.png > > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[
https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646343#comment-16646343
]
Maximilian Michels commented on BEAM-5713:
--
Actually, this is a FlinkRunner bug. The Impulse source is a regular
{{SourceFunction}}, not a {{ParallelSourceFunction}}. So the scheduler always
prefers to schedule on the node where the source gets deployed (to avoid
network traffic). After I change it to {{ParallelSourceFunction}}, the
scheduling is as expected.
> Flink portable runner schedules all tasks of streaming job on same task
> manager
> ---
>
> Key: BEAM-5713
> URL: https://issues.apache.org/jira/browse/BEAM-5713
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Affects Versions: 2.8.0
>Reporter: Thomas Weise
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
>
> The cluster has 9 task managers and 144 task slots total. A simple streaming
> pipeline with parallelism of 8 will get all tasks scheduled on the same task
> manager, causing the host to be fully booked and the remaining cluster idle.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646325#comment-16646325 ] Maximilian Michels commented on BEAM-5713: -- This is a "feature". See http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Distributing-Tasks-over-Task-manager-td9481.html and FLINK-1003. > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (BEAM-5713) Flink portable runner schedules all tasks of streaming job on same task manager
[ https://issues.apache.org/jira/browse/BEAM-5713?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels reassigned BEAM-5713: Assignee: Maximilian Michels > Flink portable runner schedules all tasks of streaming job on same task > manager > --- > > Key: BEAM-5713 > URL: https://issues.apache.org/jira/browse/BEAM-5713 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.8.0 >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > > The cluster has 9 task managers and 144 task slots total. A simple streaming > pipeline with parallelism of 8 will get all tasks scheduled on the same task > manager, causing the host to be fully booked and the remaining cluster idle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5176) FailOnWarnings behave differently between CLI and Intellij build
[
https://issues.apache.org/jira/browse/BEAM-5176?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16646226#comment-16646226
]
Maximilian Michels commented on BEAM-5176:
--
Great! Thanks for looking into this. It appears to me the
{{defaultLintSuppressions}} in {{BeamModulePlugin}} are not being passed on to
the build args. {{-Werror}} on the other hand is and the build works if it is
removed.
> FailOnWarnings behave differently between CLI and Intellij build
> -
>
> Key: BEAM-5176
> URL: https://issues.apache.org/jira/browse/BEAM-5176
> Project: Beam
> Issue Type: Sub-task
> Components: build-system
>Reporter: Etienne Chauchot
>Assignee: Kenneth Knowles
>Priority: Major
>
> In command line the build passes but fails on the IDE because of warnings.
> To make it pass I had to put false in failOnWarnings in ApplyJavaNature
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5687) Checkpointing in portable pipelines does not work
Maximilian Michels created BEAM-5687:
Summary: Checkpointing in portable pipelines does not work
Key: BEAM-5687
URL: https://issues.apache.org/jira/browse/BEAM-5687
Project: Beam
Issue Type: Bug
Components: runner-flink
Reporter: Maximilian Michels
Assignee: Maximilian Michels
Fix For: 2.9.0
Checkpoints fail:
{noformat}
AsynchronousException{java.lang.Exception: Could not materialize checkpoint 2
for operator Source: Custom Source -> 9TestInput.None/beam:env:docker:v1:0 ->
ToKeyedWorkItem (1/1).}
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointExceptionHandler.tryHandleCheckpointException(StreamTask.java:1154)
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.handleExecutionException(StreamTask.java:948)
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:885)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.Exception: Could not materialize checkpoint 2 for operator
Source: Custom Source -> 9TestInput.None/beam:env:docker:v1:0 ->
ToKeyedWorkItem (1/1).
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.handleExecutionException(StreamTask.java:943)
... 6 more
Caused by: java.util.concurrent.ExecutionException:
java.lang.NullPointerException
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at
org.apache.flink.util.FutureUtil.runIfNotDoneAndGet(FutureUtil.java:53)
at
org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.(OperatorSnapshotFinalizer.java:53)
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:854)
... 5 more
Caused by: java.lang.NullPointerException
at
org.apache.beam.runners.flink.translation.types.CoderTypeSerializer$CoderTypeSerializerConfigSnapshot.(CoderTypeSerializer.java:162)
at
org.apache.beam.runners.flink.translation.types.CoderTypeSerializer.snapshotConfiguration(CoderTypeSerializer.java:136)
at
org.apache.flink.runtime.state.RegisteredOperatorBackendStateMetaInfo.snapshot(RegisteredOperatorBackendStateMetaInfo.java:93)
at
org.apache.flink.runtime.state.DefaultOperatorStateBackend$1.performOperation(DefaultOperatorStateBackend.java:394)
at
org.apache.flink.runtime.state.DefaultOperatorStateBackend$1.performOperation(DefaultOperatorStateBackend.java:352)
at
org.apache.flink.runtime.io.async.AbstractAsyncCallableWithResources.call(AbstractAsyncCallableWithResources.java:75)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
org.apache.flink.util.FutureUtil.runIfNotDoneAndGet(FutureUtil.java:50)
... 7 more
{noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Issue Comment Deleted] (BEAM-5664) A canceled pipeline should not return a done status in the jobserver.
[ https://issues.apache.org/jira/browse/BEAM-5664?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5664: - Comment: was deleted (was: Is this still being worked on?) > A canceled pipeline should not return a done status in the jobserver. > - > > Key: BEAM-5664 > URL: https://issues.apache.org/jira/browse/BEAM-5664 > Project: Beam > Issue Type: Improvement > Components: runner-core >Reporter: Robert Bradshaw >Assignee: Kenneth Knowles >Priority: Major > Labels: portability-flink > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-3676) FlinkRunner: Portable state service
[ https://issues.apache.org/jira/browse/BEAM-3676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16641810#comment-16641810 ] Maximilian Michels commented on BEAM-3676: -- Is this still being worked on? [~thw] We can track the state of the Flink integration in BEAM-2918. > FlinkRunner: Portable state service > --- > > Key: BEAM-3676 > URL: https://issues.apache.org/jira/browse/BEAM-3676 > Project: Beam > Issue Type: Sub-task > Components: runner-flink >Reporter: Ben Sidhom >Assignee: Axel Magnuson >Priority: Major > Labels: portability, portability-flink > Time Spent: 3h 50m > Remaining Estimate: 0h > > The State API is an implementation of BeamFnState that exposes pipeline state > to SDK harnesses. Because it is used for side inputs, this service will also > need to be tied into side inputs/outputs during the translation phase. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5664) A canceled pipeline should not return a done status in the jobserver.
[ https://issues.apache.org/jira/browse/BEAM-5664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16641802#comment-16641802 ] Maximilian Michels commented on BEAM-5664: -- Is this still being worked on? > A canceled pipeline should not return a done status in the jobserver. > - > > Key: BEAM-5664 > URL: https://issues.apache.org/jira/browse/BEAM-5664 > Project: Beam > Issue Type: Improvement > Components: runner-core >Reporter: Robert Bradshaw >Assignee: Kenneth Knowles >Priority: Major > Labels: portability-flink > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5442) PortableRunner swallows custom options for Runner
[
https://issues.apache.org/jira/browse/BEAM-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels resolved BEAM-5442.
--
Resolution: Fixed
Great! I will get in contact with you to see why the checkpointing is not
working.
> PortableRunner swallows custom options for Runner
> -
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
> Time Spent: 6h 20m
> Remaining Estimate: 0h
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5668) [beam_PostCommit_Python_Verify] [:pythonPostCommit] Many tests failing
[ https://issues.apache.org/jira/browse/BEAM-5668?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5668. -- Resolution: Fixed Fix Version/s: Not applicable PostCommit is back to normal. > [beam_PostCommit_Python_Verify] [:pythonPostCommit] Many tests failing > -- > > Key: BEAM-5668 > URL: https://issues.apache.org/jira/browse/BEAM-5668 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Scott Wegner >Assignee: Maximilian Michels >Priority: Major > Labels: currently-failing > Fix For: Not applicable > > Time Spent: 2h > Remaining Estimate: 0h > > _Use this form to file an issue for test failure:_ > * [Jenkins > Job|https://builds.apache.org/job/beam_PostCommit_Python_Verify/6181/] > * [Gradle Build > Scan|https://scans.gradle.com/s/fcpta7s7wcsnm/console-log?task=:beam-sdks-python:postCommitITTests] > * [Test source > code|https://github.com/apache/beam/blob/4c693e7aa1ba71fa83370dd383f4bcebe4690b86/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py#L171] > * [Dataflow > job|https://pantheon.corp.google.com/dataflow/jobsDetail/locations/us-central1/jobs/2018-10-05_07_00_20-5526009939236014896?project=apache-beam-testing] > Initial investigation: > Lots of different looking test failures from the Gradle build scan. > > _After you've filled out the above details, please [assign the issue to an > individual|https://beam.apache.org/contribute/postcommits-guides/index.html#find_specialist]. > Assignee should [treat test failures as > high-priority|https://beam.apache.org/contribute/postcommits-policies/#assigned-failing-test], > helping to fix the issue or find a more appropriate owner. See [Apache Beam > Post-Commit > Policies|https://beam.apache.org/contribute/postcommits-policies]._ -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5668) [beam_PostCommit_Python_Verify] [:pythonPostCommit] Many tests failing
[ https://issues.apache.org/jira/browse/BEAM-5668?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16640685#comment-16640685 ] Maximilian Michels commented on BEAM-5668: -- Thanks [~swegner]. Yes, the error you posted is related to the changes. The same option is registered twice. Will make sure to run PostCommit for such changes (only ran Portable Runner tests). > [beam_PostCommit_Python_Verify] [:pythonPostCommit] Many tests failing > -- > > Key: BEAM-5668 > URL: https://issues.apache.org/jira/browse/BEAM-5668 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Scott Wegner >Assignee: Maximilian Michels >Priority: Major > Labels: currently-failing > Time Spent: 1h 40m > Remaining Estimate: 0h > > _Use this form to file an issue for test failure:_ > * [Jenkins > Job|https://builds.apache.org/job/beam_PostCommit_Python_Verify/6181/] > * [Gradle Build > Scan|https://scans.gradle.com/s/fcpta7s7wcsnm/console-log?task=:beam-sdks-python:postCommitITTests] > * [Test source > code|https://github.com/apache/beam/blob/4c693e7aa1ba71fa83370dd383f4bcebe4690b86/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py#L171] > * [Dataflow > job|https://pantheon.corp.google.com/dataflow/jobsDetail/locations/us-central1/jobs/2018-10-05_07_00_20-5526009939236014896?project=apache-beam-testing] > Initial investigation: > Lots of different looking test failures from the Gradle build scan. > > _After you've filled out the above details, please [assign the issue to an > individual|https://beam.apache.org/contribute/postcommits-guides/index.html#find_specialist]. > Assignee should [treat test failures as > high-priority|https://beam.apache.org/contribute/postcommits-policies/#assigned-failing-test], > helping to fix the issue or find a more appropriate owner. See [Apache Beam > Post-Commit > Policies|https://beam.apache.org/contribute/postcommits-policies]._ -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5664) A canceled pipeline should not return a done status in the jobserver.
[
https://issues.apache.org/jira/browse/BEAM-5664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16640007#comment-16640007
]
Maximilian Michels commented on BEAM-5664:
--
Cancel is not supported via the {{FlinkRunnerResult}}, so I'm assuming
cancellation was done through Flink itself. If it gets cancelled, an exception
should be thrown which leads to triggering JobState {{FAILED}}.
There is also {{FlinkJobInvocation#cancel()}}, this will internally throw an
exception and then leave the state unmodified, e.g. default is STOP which
should be changed to UNSPECIFIED.
{{DONE}} should only be set when the entire pipeline shuts down gracefully.
Need to dig deeper how it can be set to this state when cancelling.
> A canceled pipeline should not return a done status in the jobserver.
> -
>
> Key: BEAM-5664
> URL: https://issues.apache.org/jira/browse/BEAM-5664
> Project: Beam
> Issue Type: Improvement
> Components: runner-core
>Reporter: Robert Bradshaw
>Assignee: Kenneth Knowles
>Priority: Major
> Labels: portability-flink
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5369) Portable wordcount java broken because of create_view usage
[ https://issues.apache.org/jira/browse/BEAM-5369?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5369. -- Resolution: Fixed > Portable wordcount java broken because of create_view usage > --- > > Key: BEAM-5369 > URL: https://issues.apache.org/jira/browse/BEAM-5369 > Project: Beam > Issue Type: Test > Components: sdk-java-core >Reporter: Ankur Goenka >Assignee: Ankur Goenka >Priority: Major > Labels: portability, portability-flink > Fix For: 2.8.0 > > > Portable Wordcount is broken on Flink since > [https://github.com/apache/beam/pull/6208/files#diff-14d60e038b469ee6ce66ec6ec9d7d976L153] > SDK should stop using create_view URN as its depricated -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5528) Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
[
https://issues.apache.org/jira/browse/BEAM-5528?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels resolved BEAM-5528.
--
Resolution: Fixed
> Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
>
>
> Key: BEAM-5528
> URL: https://issues.apache.org/jira/browse/BEAM-5528
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: portability-flink
> Fix For: 2.8.0
>
> Time Spent: 2h 20m
> Remaining Estimate: 0h
>
> We recently removed the CREATE_VIEW translation which was a NOOP. Turns out
> we still need to support it until it is removed.
> {noformat}
> java.lang.IllegalArgumentException: Unknown type of URN
> beam:transform:create_view:v1 for PTransform with id
> TextIO.Write/WriteFiles/GatherTempFileResults/View.AsList/View.CreatePCollectionView.
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.urnNotFound(FlinkBatchPortablePipelineTranslator.java:578)
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:233)
> at
> org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:112)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5649) Remove deprecated primitive CREATE_VIEW transform from Runner API
[
https://issues.apache.org/jira/browse/BEAM-5649?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16639559#comment-16639559
]
Maximilian Michels commented on BEAM-5649:
--
Another solution that we discussed in
https://github.com/apache/beam/pull/6523#issuecomment-427160991 was to
construct a different graph for the portable pipelines. This has the drawback
that the graphs generation logic can't be shared across non-portable and
portable pipelines.
> Remove deprecated primitive CREATE_VIEW transform from Runner API
> -
>
> Key: BEAM-5649
> URL: https://issues.apache.org/jira/browse/BEAM-5649
> Project: Beam
> Issue Type: Improvement
> Components: sdk-java-core
>Reporter: Maximilian Michels
>Priority: Minor
> Labels: portability
>
> The deprecated {{CREATE_VIEW}} transform is still generated as part of the
> Java SDK pipeline construction but not as part of the Python SDK. We had
> previously removed support for the transform in the Portable FlinkRunner
> because end-to-end tests were only run with Python. Since the Java SDK still
> generates the transform we've re-added support for it but would like to
> eventually get it removed.
> This issue tracks removal of the transform from the Runner API. The transform
> can stay in the non-portable Runner pipeline translation.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (BEAM-5649) Remove deprecated primitive CREATE_VIEW transform from Runner API
[
https://issues.apache.org/jira/browse/BEAM-5649?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels reassigned BEAM-5649:
Assignee: (was: Maximilian Michels)
> Remove deprecated primitive CREATE_VIEW transform from Runner API
> -
>
> Key: BEAM-5649
> URL: https://issues.apache.org/jira/browse/BEAM-5649
> Project: Beam
> Issue Type: Improvement
> Components: sdk-java-core
>Reporter: Maximilian Michels
>Priority: Minor
> Labels: portability
>
> The deprecated {{CREATE_VIEW}} transform is still generated as part of the
> Java SDK pipeline construction but not as part of the Python SDK. We had
> previously removed support for the transform in the Portable FlinkRunner
> because end-to-end tests were only run with Python. Since the Java SDK still
> generates the transform we've re-added support for it but would like to
> eventually get it removed.
> This issue tracks removal of the transform from the Runner API. The transform
> can stay in the non-portable Runner pipeline translation.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5649) Remove deprecated primitive CREATE_VIEW transform from Runner API
[
https://issues.apache.org/jira/browse/BEAM-5649?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638639#comment-16638639
]
Maximilian Michels commented on BEAM-5649:
--
The translation code of the non-portable Runners is also tight to the URNs of
the Runner API. It looks like removing the transform from the Runner API also
requires removing it from the old Runners.
> Remove deprecated primitive CREATE_VIEW transform from Runner API
> -
>
> Key: BEAM-5649
> URL: https://issues.apache.org/jira/browse/BEAM-5649
> Project: Beam
> Issue Type: Improvement
> Components: sdk-java-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Minor
> Labels: portability
>
> The deprecated {{CREATE_VIEW}} transform is still generated as part of the
> Java SDK pipeline construction but not as part of the Python SDK. We had
> previously removed support for the transform in the Portable FlinkRunner
> because end-to-end tests were only run with Python. Since the Java SDK still
> generates the transform we've re-added support for it but would like to
> eventually get it removed.
> This issue tracks removal of the transform from the Runner API. The transform
> can stay in the non-portable Runner pipeline translation.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5649) Remove deprecated primitive CREATE_VIEW transform from Runner API
Maximilian Michels created BEAM-5649:
Summary: Remove deprecated primitive CREATE_VIEW transform from
Runner API
Key: BEAM-5649
URL: https://issues.apache.org/jira/browse/BEAM-5649
Project: Beam
Issue Type: Improvement
Components: sdk-java-core
Reporter: Maximilian Michels
Assignee: Maximilian Michels
The deprecated {{CREATE_VIEW}} transform is still generated as part of the Java
SDK pipeline construction but not as part of the Python SDK. We had previously
removed support for the transform in the Portable FlinkRunner because
end-to-end tests were only run with Python. Since the Java SDK still generates
the transform we've re-added support for it but would like to eventually get it
removed.
This issue tracks removal of the transform from the Runner API. The transform
can stay in the non-portable Runner pipeline translation.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-4819) Make portable Flink runner JobBundleFactory configurable
[ https://issues.apache.org/jira/browse/BEAM-4819?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638542#comment-16638542 ] Maximilian Michels commented on BEAM-4819: -- That's what I wanted to find out. So we keep this issue open. > Make portable Flink runner JobBundleFactory configurable > > > Key: BEAM-4819 > URL: https://issues.apache.org/jira/browse/BEAM-4819 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Major > Labels: portability, portability-flink > > BEAM-4791 introduces factory override for testing, expand that to allow users > to configure a different factory via service loader to adopt alternative > execution environments. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5419) Build multiple versions of the Flink Runner against different Flink versions
[ https://issues.apache.org/jira/browse/BEAM-5419?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638367#comment-16638367 ] Maximilian Michels commented on BEAM-5419: -- Won't be fixed for 2.8.0. > Build multiple versions of the Flink Runner against different Flink versions > > > Key: BEAM-5419 > URL: https://issues.apache.org/jira/browse/BEAM-5419 > Project: Beam > Issue Type: New Feature > Components: build-system, runner-flink >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Major > Fix For: 2.9.0 > > > Following up on a discussion on the mailing list. > We want to keep the Flink version stable across different versions to avoid > upgrade pain for long-term users. At the same time, there are users out there > with newer Flink clusters and developers also want to utilize new Flink > features. > It would be great to build multiple versions of the Flink Runner against > different Flink versions. > When the upgrade is as simple as changing the version property in the build > script, this should be pretty straight-forward. If not, having a "base > version" and applying a patch during the build could be an option. We should > avoid duplicating any Runner code. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5528) Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
[
https://issues.apache.org/jira/browse/BEAM-5528?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638379#comment-16638379
]
Maximilian Michels commented on BEAM-5528:
--
Just discovered this is already reported via BEAM-5369.
> Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
>
>
> Key: BEAM-5528
> URL: https://issues.apache.org/jira/browse/BEAM-5528
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: portability-flink
> Fix For: 2.8.0
>
> Time Spent: 1h
> Remaining Estimate: 0h
>
> We recently removed the CREATE_VIEW translation which was a NOOP. Turns out
> we still need to support it until it is removed.
> {noformat}
> java.lang.IllegalArgumentException: Unknown type of URN
> beam:transform:create_view:v1 for PTransform with id
> TextIO.Write/WriteFiles/GatherTempFileResults/View.AsList/View.CreatePCollectionView.
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.urnNotFound(FlinkBatchPortablePipelineTranslator.java:578)
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:233)
> at
> org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:112)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5369) Portable wordcount java broken because of create_view usage
[ https://issues.apache.org/jira/browse/BEAM-5369?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5369: - Labels: portability portability-flink (was: ) > Portable wordcount java broken because of create_view usage > --- > > Key: BEAM-5369 > URL: https://issues.apache.org/jira/browse/BEAM-5369 > Project: Beam > Issue Type: Test > Components: sdk-java-core >Reporter: Ankur Goenka >Assignee: Ankur Goenka >Priority: Major > Labels: portability, portability-flink > Fix For: 2.8.0 > > > Portable Wordcount is broken on Flink since > [https://github.com/apache/beam/pull/6208/files#diff-14d60e038b469ee6ce66ec6ec9d7d976L153] > SDK should stop using create_view URN as its depricated -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5369) Portable wordcount java broken because of create_view usage
[ https://issues.apache.org/jira/browse/BEAM-5369?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5369: - Fix Version/s: 2.8.0 > Portable wordcount java broken because of create_view usage > --- > > Key: BEAM-5369 > URL: https://issues.apache.org/jira/browse/BEAM-5369 > Project: Beam > Issue Type: Test > Components: sdk-java-core >Reporter: Ankur Goenka >Assignee: Ankur Goenka >Priority: Major > Labels: portability, portability-flink > Fix For: 2.8.0 > > > Portable Wordcount is broken on Flink since > [https://github.com/apache/beam/pull/6208/files#diff-14d60e038b469ee6ce66ec6ec9d7d976L153] > SDK should stop using create_view URN as its depricated -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5419) Build multiple versions of the Flink Runner against different Flink versions
[ https://issues.apache.org/jira/browse/BEAM-5419?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5419: - Fix Version/s: (was: 2.8.0) 2.9.0 > Build multiple versions of the Flink Runner against different Flink versions > > > Key: BEAM-5419 > URL: https://issues.apache.org/jira/browse/BEAM-5419 > Project: Beam > Issue Type: New Feature > Components: build-system, runner-flink >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Major > Fix For: 2.9.0 > > > Following up on a discussion on the mailing list. > We want to keep the Flink version stable across different versions to avoid > upgrade pain for long-term users. At the same time, there are users out there > with newer Flink clusters and developers also want to utilize new Flink > features. > It would be great to build multiple versions of the Flink Runner against > different Flink versions. > When the upgrade is as simple as changing the version property in the build > script, this should be pretty straight-forward. If not, having a "base > version" and applying a patch during the build could be an option. We should > avoid duplicating any Runner code. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5267) Update Flink Runner to Flink 1.6.x
[ https://issues.apache.org/jira/browse/BEAM-5267?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5267: - Fix Version/s: (was: 2.8.0) 2.9.0 > Update Flink Runner to Flink 1.6.x > -- > > Key: BEAM-5267 > URL: https://issues.apache.org/jira/browse/BEAM-5267 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Major > Fix For: 2.9.0 > > Time Spent: 40m > Remaining Estimate: 0h > > For the next release, the Flink version should be bumped. As changes for > 2.7.0 are already frozen, it's going to be 2.8.0. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5267) Update Flink Runner to Flink 1.6.x
[ https://issues.apache.org/jira/browse/BEAM-5267?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638366#comment-16638366 ] Maximilian Michels commented on BEAM-5267: -- This depends on BEAM-5419 and won't be fixed for 2.8.0. > Update Flink Runner to Flink 1.6.x > -- > > Key: BEAM-5267 > URL: https://issues.apache.org/jira/browse/BEAM-5267 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Major > Fix For: 2.9.0 > > Time Spent: 40m > Remaining Estimate: 0h > > For the next release, the Flink version should be bumped. As changes for > 2.7.0 are already frozen, it's going to be 2.8.0. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-4819) Make portable Flink runner JobBundleFactory configurable
[ https://issues.apache.org/jira/browse/BEAM-4819?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638276#comment-16638276 ] Maximilian Michels commented on BEAM-4819: -- Is this still an issue after https://github.com/apache/beam/pull/6441 which enables to configure the environment (docker/process/embedded)? I think that was the main motivation for this issue. > Make portable Flink runner JobBundleFactory configurable > > > Key: BEAM-4819 > URL: https://issues.apache.org/jira/browse/BEAM-4819 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Major > Labels: portability, portability-flink > > BEAM-4791 introduces factory override for testing, expand that to allow users > to configure a different factory via service loader to adopt alternative > execution environments. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5528) Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
[
https://issues.apache.org/jira/browse/BEAM-5528?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16638046#comment-16638046
]
Maximilian Michels commented on BEAM-5528:
--
Yes, this should go into 2.8.0. Pull request is open.
> Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
>
>
> Key: BEAM-5528
> URL: https://issues.apache.org/jira/browse/BEAM-5528
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: portability-flink
> Fix For: 2.8.0
>
> Time Spent: 1h
> Remaining Estimate: 0h
>
> We recently removed the CREATE_VIEW translation which was a NOOP. Turns out
> we still need to support it until it is removed.
> {noformat}
> java.lang.IllegalArgumentException: Unknown type of URN
> beam:transform:create_view:v1 for PTransform with id
> TextIO.Write/WriteFiles/GatherTempFileResults/View.AsList/View.CreatePCollectionView.
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.urnNotFound(FlinkBatchPortablePipelineTranslator.java:578)
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:233)
> at
> org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:112)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5528) Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
[
https://issues.apache.org/jira/browse/BEAM-5528?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5528:
-
Priority: Critical (was: Major)
> Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
>
>
> Key: BEAM-5528
> URL: https://issues.apache.org/jira/browse/BEAM-5528
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: portability-flink
> Fix For: 2.8.0
>
> Time Spent: 1h
> Remaining Estimate: 0h
>
> We recently removed the CREATE_VIEW translation which was a NOOP. Turns out
> we still need to support it until it is removed.
> {noformat}
> java.lang.IllegalArgumentException: Unknown type of URN
> beam:transform:create_view:v1 for PTransform with id
> TextIO.Write/WriteFiles/GatherTempFileResults/View.AsList/View.CreatePCollectionView.
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.urnNotFound(FlinkBatchPortablePipelineTranslator.java:578)
> at
> org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:233)
> at
> org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:112)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5440) Add option to mount a directory inside SDK harness containers
[
https://issues.apache.org/jira/browse/BEAM-5440?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5440:
-
Fix Version/s: 2.9.0
> Add option to mount a directory inside SDK harness containers
> -
>
> Key: BEAM-5440
> URL: https://issues.apache.org/jira/browse/BEAM-5440
> Project: Beam
> Issue Type: New Feature
> Components: java-fn-execution
>Reporter: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.9.0
>
>
> While experimenting with the Python SDK locally, I found it inconvenient that
> I can't mount a host directory to the Docker containers, i.e. the input must
> already be in the container and the results of a Write remain inside the
> container. For local testing, users may want to mount a host directory.
> Since BEAM-5288 the {{Environment}} carries explicit environment information,
> we could a) add volume args to the {{DockerPayload}}, or b) provide a general
> Docker arguments field.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5440) Add option to mount a directory inside SDK harness containers
[
https://issues.apache.org/jira/browse/BEAM-5440?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16637980#comment-16637980
]
Maximilian Michels commented on BEAM-5440:
--
[~altay] Not a blocker. I generally always provide a Fix Version. A common
practice is to bulk move unresolved issues to the next version upon release. So
I would mark this as 2.9.0 now.
> Add option to mount a directory inside SDK harness containers
> -
>
> Key: BEAM-5440
> URL: https://issues.apache.org/jira/browse/BEAM-5440
> Project: Beam
> Issue Type: New Feature
> Components: java-fn-execution
>Reporter: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.9.0
>
>
> While experimenting with the Python SDK locally, I found it inconvenient that
> I can't mount a host directory to the Docker containers, i.e. the input must
> already be in the container and the results of a Write remain inside the
> container. For local testing, users may want to mount a host directory.
> Since BEAM-5288 the {{Environment}} carries explicit environment information,
> we could a) add volume args to the {{DockerPayload}}, or b) provide a general
> Docker arguments field.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5442) PortableRunner swallows custom options for Runner
[
https://issues.apache.org/jira/browse/BEAM-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16637961#comment-16637961
]
Maximilian Michels commented on BEAM-5442:
--
Yes, this should go into 2.8.0. [~altay] You can check out the PR if you're
interested: https://github.com/apache/beam/pull/6557
> PortableRunner swallows custom options for Runner
> -
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5187) Create a ProcessJobBundleFactory for non-dockerized SDK harness
[
https://issues.apache.org/jira/browse/BEAM-5187?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels resolved BEAM-5187.
--
Resolution: Implemented
Fix Version/s: 2.7.0
Artifact staging is not integrated but [~thw] has tried it out and stated it is
not required for his use case. We might follow up with a programmatic way to
stage artifacts. Closing this for now since the basic process-based execution
is in place.
> Create a ProcessJobBundleFactory for non-dockerized SDK harness
> ---
>
> Key: BEAM-5187
> URL: https://issues.apache.org/jira/browse/BEAM-5187
> Project: Beam
> Issue Type: New Feature
> Components: runner-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Minor
> Fix For: 2.7.0
>
> Time Spent: 6h 40m
> Remaining Estimate: 0h
>
> As discussed on the mailing list [1], we want to giver users an option to
> execute portable pipelines without Docker. Analog to the
> {{DockerJobBundleFactory}}, a {{ProcessJobBundleFactory}} could be added to
> directly fork SDK harness processes.
> Artifacts will be provided by an artifact directory or could be setup similar
> to the existing bootstrapping code ("boot.go") which we use for containers.
> The process-based execution can optionally be configured via the pipeline
> options.
> [1]
> [https://lists.apache.org/thread.html/d8b81e9f74f77d74c8b883cda80fa48efdcaf6ac2ad313c4fe68795a@%3Cdev.beam.apache.org%3E]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Closed] (BEAM-5527) Java PortableRunner produces wrongly shaded jar
[
https://issues.apache.org/jira/browse/BEAM-5527?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels closed BEAM-5527.
Resolution: Duplicate
Fix Version/s: (was: 2.8.0)
Not applicable
> Java PortableRunner produces wrongly shaded jar
> ---
>
> Key: BEAM-5527
> URL: https://issues.apache.org/jira/browse/BEAM-5527
> Project: Beam
> Issue Type: Bug
> Components: runner-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: Not applicable
>
>
> Was trying out the Java Portable Runner. {{beam-runners-reference-java}} is
> shaded but does not contain the required Guava classes:
> {noformat}
> Exception in thread "main" java.lang.NoClassDefFoundError:
> org/apache/beam/repackaged/beam_runners_reference_java/com/google/common/collect/ImmutableList
> at
> org.apache.beam.runners.reference.PortableRunnerRegistrar.getPipelineRunners(PortableRunnerRegistrar.java:31)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory$Cache.(PipelineOptionsFactory.java:1705)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory$Cache.(PipelineOptionsFactory.java:1679)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.resetCache(PipelineOptionsFactory.java:539)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.(PipelineOptionsFactory.java:505)
> at PortableWordCount.main(PortableWordCount.java:65)
> Caused by: java.lang.ClassNotFoundException:
> org.apache.beam.repackaged.beam_runners_reference_java.com.google.common.collect.ImmutableList
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> ... 6 more
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5528) Java PortableRunner pipeline fails on FlinkRunner due to CREATE_VIEW
Maximilian Michels created BEAM-5528:
Summary: Java PortableRunner pipeline fails on FlinkRunner due to
CREATE_VIEW
Key: BEAM-5528
URL: https://issues.apache.org/jira/browse/BEAM-5528
Project: Beam
Issue Type: Bug
Components: runner-flink
Reporter: Maximilian Michels
Assignee: Maximilian Michels
Fix For: 2.8.0
We recently removed the CREATE_VIEW translation which was a NOOP. Turns out we
still need to support it until it is removed.
{noformat}
java.lang.IllegalArgumentException: Unknown type of URN
beam:transform:create_view:v1 for PTransform with id
TextIO.Write/WriteFiles/GatherTempFileResults/View.AsList/View.CreatePCollectionView.
at
org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.urnNotFound(FlinkBatchPortablePipelineTranslator.java:578)
at
org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:233)
at
org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:112)
at
org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
at
org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
at
org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
{noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5527) Java PortableRunner produces wrongly shaded jar
Maximilian Michels created BEAM-5527:
Summary: Java PortableRunner produces wrongly shaded jar
Key: BEAM-5527
URL: https://issues.apache.org/jira/browse/BEAM-5527
Project: Beam
Issue Type: Bug
Components: runner-core
Reporter: Maximilian Michels
Assignee: Maximilian Michels
Fix For: 2.8.0
Was trying out the Java Portable Runner. {{beam-runners-reference-java}} is
shaded but does not contain the required Guava classes:
{noformat}
Exception in thread "main" java.lang.NoClassDefFoundError:
org/apache/beam/repackaged/beam_runners_reference_java/com/google/common/collect/ImmutableList
at
org.apache.beam.runners.reference.PortableRunnerRegistrar.getPipelineRunners(PortableRunnerRegistrar.java:31)
at
org.apache.beam.sdk.options.PipelineOptionsFactory$Cache.(PipelineOptionsFactory.java:1705)
at
org.apache.beam.sdk.options.PipelineOptionsFactory$Cache.(PipelineOptionsFactory.java:1679)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.resetCache(PipelineOptionsFactory.java:539)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.(PipelineOptionsFactory.java:505)
at PortableWordCount.main(PortableWordCount.java:65)
Caused by: java.lang.ClassNotFoundException:
org.apache.beam.repackaged.beam_runners_reference_java.com.google.common.collect.ImmutableList
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 6 more
{noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5458) Python PreCommit broken
[ https://issues.apache.org/jira/browse/BEAM-5458?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5458. -- Resolution: Fixed Assignee: Robbe Fix Version/s: Not applicable > Python PreCommit broken > --- > > Key: BEAM-5458 > URL: https://issues.apache.org/jira/browse/BEAM-5458 > Project: Beam > Issue Type: Test > Components: test-failures, testing >Reporter: Maximilian Michels >Assignee: Robbe >Priority: Critical > Fix For: Not applicable > > Time Spent: 1h > Remaining Estimate: 0h > > Python PreCommit is currently broken: > https://builds.apache.org/job/beam_PreCommit_Python_Commit/1308/#showFailuresLink -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-5458) Python PreCommit broken
Maximilian Michels created BEAM-5458: Summary: Python PreCommit broken Key: BEAM-5458 URL: https://issues.apache.org/jira/browse/BEAM-5458 Project: Beam Issue Type: Test Components: testing Reporter: Maximilian Michels Python PreCommit is currently broken: https://builds.apache.org/job/beam_PreCommit_Python_Commit/1308/#showFailuresLink -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5444) Post Commit Failure in Flink Runner
[ https://issues.apache.org/jira/browse/BEAM-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5444. -- Resolution: Fixed Fix Version/s: Not applicable > Post Commit Failure in Flink Runner > --- > > Key: BEAM-5444 > URL: https://issues.apache.org/jira/browse/BEAM-5444 > Project: Beam > Issue Type: Bug > Components: runner-flink, test-failures >Reporter: Robert Burke >Assignee: Maximilian Michels >Priority: Major > Fix For: Not applicable > > Time Spent: 1h 10m > Remaining Estimate: 0h > > The first instance of this failure included entirely your commits, could you > investigate? > [https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Gradle/1544/] > It appears expected exceptions are either being swallowed and not thrown. > SEVERE: Pipeline execution failed > org.apache.flink.runtime.client.JobExecutionException: java.lang.Exception: > The user defined 'open()' method caused an exception: java.lang.Exception: > Hasn't yet thrown -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5444) Post Commit Failure in Flink Runner
[ https://issues.apache.org/jira/browse/BEAM-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622529#comment-16622529 ] Maximilian Michels commented on BEAM-5444: -- The default parallelism for local execution changed from 1 to number of cpu cores. This breaks the test. Opening a PR. > Post Commit Failure in Flink Runner > --- > > Key: BEAM-5444 > URL: https://issues.apache.org/jira/browse/BEAM-5444 > Project: Beam > Issue Type: Bug > Components: runner-flink, test-failures >Reporter: Robert Burke >Assignee: Maximilian Michels >Priority: Major > > The first instance of this failure included entirely your commits, could you > investigate? > [https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Gradle/1544/] > It appears expected exceptions are either being swallowed and not thrown. > SEVERE: Pipeline execution failed > org.apache.flink.runtime.client.JobExecutionException: java.lang.Exception: > The user defined 'open()' method caused an exception: java.lang.Exception: > Hasn't yet thrown -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5442) PortableRunner swallows custom options for Runner
[
https://issues.apache.org/jira/browse/BEAM-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622500#comment-16622500
]
Maximilian Michels commented on BEAM-5442:
--
There are some pros to doing an upfront parsing/validation of the options.
Maybe both can work complementary. The options can be parsed/validated but
unknown options should still be forwarded.
> PortableRunner swallows custom options for Runner
> -
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
> Time Spent: 10m
> Remaining Estimate: 0h
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5442) PortableRunner swallows custom options for Runner
[
https://issues.apache.org/jira/browse/BEAM-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622449#comment-16622449
]
Maximilian Michels commented on BEAM-5442:
--
Actually, if you look at {{get_all_options}} method you see it filter out all
the options which are not part of a {{PipelineOptions}} class. See
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/options/pipeline_options.py#L212
So we just need to add a FlinkOptions derived from PipelineOptions to the
Python SDK.
> PortableRunner swallows custom options for Runner
> -
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (BEAM-5442) PortableRunner swallows custom options for Runner
[
https://issues.apache.org/jira/browse/BEAM-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels reassigned BEAM-5442:
Assignee: Maximilian Michels
> PortableRunner swallows custom options for Runner
> -
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5288) Modify Environment to support non-dockerized SDK harness deployments
[
https://issues.apache.org/jira/browse/BEAM-5288?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5288:
-
Labels: (was: por)
> Modify Environment to support non-dockerized SDK harness deployments
> -
>
> Key: BEAM-5288
> URL: https://issues.apache.org/jira/browse/BEAM-5288
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
>Reporter: Maximilian Michels
>Assignee: Ankur Goenka
>Priority: Major
> Time Spent: 5h
> Remaining Estimate: 0h
>
> As of mailing discussions and BEAM-5187, it has become clear that we need to
> extend the Environment information. In addition to the Docker environment,
> the extended environment holds deployment options for 1) a process-based
> environment, 2) an externally managed environment.
> The proto definition, as of now, looks as follows:
> {noformat}
> message Environment {
>// (Required) The URN of the payload
>string urn = 1;
>// (Optional) The data specifying any parameters to the URN. If
>// the URN does not require any arguments, this may be omitted.
>bytes payload = 2;
> }
> message StandardEnvironments {
>enum Environments {
> DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
> PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
> EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
>}
> }
> // The payload of a Docker image
> message DockerPayload {
>string container_image = 1; // implicitly linux_amd64.
> }
> message ProcessPayload {
>string os = 1; // "linux", "darwin", ..
>string arch = 2; // "amd64", ..
>string command = 3; // process to execute
>map env = 4; // environment variables
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5288) Modify Environment to support non-dockerized SDK harness deployments
[
https://issues.apache.org/jira/browse/BEAM-5288?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5288:
-
Labels: por (was: )
> Modify Environment to support non-dockerized SDK harness deployments
> -
>
> Key: BEAM-5288
> URL: https://issues.apache.org/jira/browse/BEAM-5288
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
>Reporter: Maximilian Michels
>Assignee: Ankur Goenka
>Priority: Major
> Time Spent: 5h
> Remaining Estimate: 0h
>
> As of mailing discussions and BEAM-5187, it has become clear that we need to
> extend the Environment information. In addition to the Docker environment,
> the extended environment holds deployment options for 1) a process-based
> environment, 2) an externally managed environment.
> The proto definition, as of now, looks as follows:
> {noformat}
> message Environment {
>// (Required) The URN of the payload
>string urn = 1;
>// (Optional) The data specifying any parameters to the URN. If
>// the URN does not require any arguments, this may be omitted.
>bytes payload = 2;
> }
> message StandardEnvironments {
>enum Environments {
> DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
> PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
> EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
>}
> }
> // The payload of a Docker image
> message DockerPayload {
>string container_image = 1; // implicitly linux_amd64.
> }
> message ProcessPayload {
>string os = 1; // "linux", "darwin", ..
>string arch = 2; // "amd64", ..
>string command = 3; // process to execute
>map env = 4; // environment variables
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5441) Portable Wordcount fails in GreedyPipelineFuser
[
https://issues.apache.org/jira/browse/BEAM-5441?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5441:
-
Labels: portability portability-flink (was: )
> Portable Wordcount fails in GreedyPipelineFuser
> ---
>
> Key: BEAM-5441
> URL: https://issues.apache.org/jira/browse/BEAM-5441
> Project: Beam
> Issue Type: New Feature
> Components: examples-python, sdk-java-core
>Affects Versions: 2.8.0
>Reporter: Maximilian Michels
>Assignee: Ahmet Altay
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
>
> The Python SDK wordcount with the PortableRunner throws the following
> exception:
> {noformat}
> java.lang.IllegalArgumentException: A PCollectionNode should have exactly one
> producing PTransformNode, PCollectionNode{id=ref_PCollection_PCollection_26,
> PCollection=unique_name:
> "60write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys.None"
> coder_id: "ref_Coder_FastPrimitivesCoder_2"
> is_bounded: BOUNDED
> windowing_strategy_id: "ref_Windowing_Windowing_1"
> } has
> [PTransformNode{id=ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle_32,
> transform=spec {
> urn: "beam:transform:generic_composite:v1"
> payload: ""
> }
> subtransforms:
> "ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/AddRandomKeys_33"
> subtransforms:
> "ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/ReshufflePerKey_34"
> subtransforms:
> "ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys_41"
> inputs {
> key: "0"
> value: "ref_PCollection_PCollection_19"
> }
> outputs {
> key: "None"
> value: "ref_PCollection_PCollection_26"
> }
> unique_name: "write/Write/WriteImpl/DoOnce/Read/Reshuffle"
> },
> PTransformNode{id=ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys_41,
> transform=spec {
> urn: "urn:beam:transform:pardo:v1"
> payload: "\n\317\006\n\255\006\n
> beam:dofn:pickled_python_info:v1\032\210\006eNrFkttP1EAUxtsFZe1yUcEL4v3aRbdFUEFFQEHUEGpSnPhkxml32Gnstv3a6YZNbKIxJf7ZTvdB2ER9NZNMcs6cb3LO73zfaqbPEuYLTj3OupZMWZTtx2k3s/w45cYmC0PmhfxjypKEp1vxdmRAa36HXqBmkrEkjX2eZRjx20EYWrS6DeqnnElO9/PIl0GsFKPm0HsYszaV/YQbOEHqm3Gbf1ABTpYYc1E3d3R1arvTG2Tip6Z91bQfutbRtT2cckoYTaIfoFFinPRtkvE0s7vswN7iPbuaoCV5Ju0ej3p2GHh20pcijhatZTsLJG+pSb+wDs/sYzO3Fq0Va8Fq895CK+mrUot3OscL7CModgXFSvqYIPXVkHW9NlvD5G5jlGiYIrX9CKdLnGlKnHUx7VPq5UEog4hSo8MlkzI1MDNEIugmcSppN27noaJxjsz9Yxs4X+KCi4ukTpXcl5Ri9hCXXMyJSedPC/C5CnBZjJriN9W9z6SukLZ1bXYPV5wd/RBXFVKJayWu/w+kuQzCCukNMbm7XhNTTYXvpotbYkb8HUclwu0Sd1zcFQrCPRemguAUaJLGwFpUBJHMMD9sb/UwyKveFFEm4zQz3r2v3Pe2Shu4r7z9oECrgGWSRhAluRx8l8F2yHicy6PEgpMf4qGXSSy6WCrxyMXjEk8KLJtiXlQfrRR4WuCZKWxHDKqfe6o7lnayhPtUuWVVLOUSL1ysDXe9PpBvFHhZ4NWRfNMjI5VsS6zl3ie8LrDtOBJvrF+Bv0km\022\035ref_Environment_Environment_1"
> }
> inputs {
> key: "0"
> value: "ref_PCollection_PCollection_25"
> }
> outputs {
> key: "None"
> value: "ref_PCollection_PCollection_26"
> }
> unique_name: "write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys"
> }]
> at
> org.apache.beam.repackaged.beam_runners_core_construction_java.com.google.common.base.Preconditions.checkArgument(Preconditions.java:416)
> at
> org.apache.beam.runners.core.construction.graph.QueryablePipeline.buildNetwork(QueryablePipeline.java:176)
> at
> org.apache.beam.runners.core.construction.graph.QueryablePipeline.(QueryablePipeline.java:119)
> at
> org.apache.beam.runners.core.construction.graph.QueryablePipeline.forPrimitivesIn(QueryablePipeline.java:82)
> at
> org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.(GreedyPipelineFuser.java:67)
> at
> org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.fuse(GreedyPipelineFuser.java:89)
> at
> org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:96)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> Looks like it was caused by
[jira] [Updated] (BEAM-5440) Add option to mount a directory inside SDK harness containers
[
https://issues.apache.org/jira/browse/BEAM-5440?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5440:
-
Labels: portability portability-flink (was: )
> Add option to mount a directory inside SDK harness containers
> -
>
> Key: BEAM-5440
> URL: https://issues.apache.org/jira/browse/BEAM-5440
> Project: Beam
> Issue Type: New Feature
> Components: java-fn-execution
>Reporter: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
>
> While experimenting with the Python SDK locally, I found it inconvenient that
> I can't mount a host directory to the Docker containers, i.e. the input must
> already be in the container and the results of a Write remain inside the
> container. For local testing, users may want to mount a host directory.
> Since BEAM-5288 the {{Environment}} carries explicit environment information,
> we could a) add volume args to the {{DockerPayload}}, or b) provide a general
> Docker arguments field.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5442) PortableRunner swallows custom options for Runner
[
https://issues.apache.org/jira/browse/BEAM-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5442:
-
Labels: portability portability-flink (was: )
> PortableRunner swallows custom options for Runner
> -
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
>Reporter: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.8.0
>
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5442) PortableRunner swallows custom options for Runner
Maximilian Michels created BEAM-5442:
Summary: PortableRunner swallows custom options for Runner
Key: BEAM-5442
URL: https://issues.apache.org/jira/browse/BEAM-5442
Project: Beam
Issue Type: Bug
Components: sdk-java-core, sdk-py-core
Reporter: Maximilian Michels
Assignee: Kenneth Knowles
Fix For: 2.8.0
The PortableRunner doesn't pass custom PipelineOptions to the executing Runner.
Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
(The option is just removed during proto translation without any warning)
We should allow some form of customization through the options, even for the
PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5441) Portable Wordcount fails in GreedyPipelineFuser
Maximilian Michels created BEAM-5441:
Summary: Portable Wordcount fails in GreedyPipelineFuser
Key: BEAM-5441
URL: https://issues.apache.org/jira/browse/BEAM-5441
Project: Beam
Issue Type: New Feature
Components: examples-python, sdk-java-core
Affects Versions: 2.8.0
Reporter: Maximilian Michels
Assignee: Ahmet Altay
Fix For: 2.8.0
The Python SDK wordcount with the PortableRunner throws the following exception:
{noformat}
java.lang.IllegalArgumentException: A PCollectionNode should have exactly one
producing PTransformNode, PCollectionNode{id=ref_PCollection_PCollection_26,
PCollection=unique_name:
"60write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys.None"
coder_id: "ref_Coder_FastPrimitivesCoder_2"
is_bounded: BOUNDED
windowing_strategy_id: "ref_Windowing_Windowing_1"
} has
[PTransformNode{id=ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle_32,
transform=spec {
urn: "beam:transform:generic_composite:v1"
payload: ""
}
subtransforms:
"ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/AddRandomKeys_33"
subtransforms:
"ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/ReshufflePerKey_34"
subtransforms:
"ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys_41"
inputs {
key: "0"
value: "ref_PCollection_PCollection_19"
}
outputs {
key: "None"
value: "ref_PCollection_PCollection_26"
}
unique_name: "write/Write/WriteImpl/DoOnce/Read/Reshuffle"
},
PTransformNode{id=ref_AppliedPTransform_write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys_41,
transform=spec {
urn: "urn:beam:transform:pardo:v1"
payload: "\n\317\006\n\255\006\n
beam:dofn:pickled_python_info:v1\032\210\006eNrFkttP1EAUxtsFZe1yUcEL4v3aRbdFUEFFQEHUEGpSnPhkxml32Gnstv3a6YZNbKIxJf7ZTvdB2ER9NZNMcs6cb3LO73zfaqbPEuYLTj3OupZMWZTtx2k3s/w45cYmC0PmhfxjypKEp1vxdmRAa36HXqBmkrEkjX2eZRjx20EYWrS6DeqnnElO9/PIl0GsFKPm0HsYszaV/YQbOEHqm3Gbf1ABTpYYc1E3d3R1arvTG2Tip6Z91bQfutbRtT2cckoYTaIfoFFinPRtkvE0s7vswN7iPbuaoCV5Ju0ej3p2GHh20pcijhatZTsLJG+pSb+wDs/sYzO3Fq0Va8Fq895CK+mrUot3OscL7CModgXFSvqYIPXVkHW9NlvD5G5jlGiYIrX9CKdLnGlKnHUx7VPq5UEog4hSo8MlkzI1MDNEIugmcSppN27noaJxjsz9Yxs4X+KCi4ukTpXcl5Ri9hCXXMyJSedPC/C5CnBZjJriN9W9z6SukLZ1bXYPV5wd/RBXFVKJayWu/w+kuQzCCukNMbm7XhNTTYXvpotbYkb8HUclwu0Sd1zcFQrCPRemguAUaJLGwFpUBJHMMD9sb/UwyKveFFEm4zQz3r2v3Pe2Shu4r7z9oECrgGWSRhAluRx8l8F2yHicy6PEgpMf4qGXSSy6WCrxyMXjEk8KLJtiXlQfrRR4WuCZKWxHDKqfe6o7lnayhPtUuWVVLOUSL1ysDXe9PpBvFHhZ4NWRfNMjI5VsS6zl3ie8LrDtOBJvrF+Bv0km\022\035ref_Environment_Environment_1"
}
inputs {
key: "0"
value: "ref_PCollection_PCollection_25"
}
outputs {
key: "None"
value: "ref_PCollection_PCollection_26"
}
unique_name: "write/Write/WriteImpl/DoOnce/Read/Reshuffle/RemoveRandomKeys"
}]
at
org.apache.beam.repackaged.beam_runners_core_construction_java.com.google.common.base.Preconditions.checkArgument(Preconditions.java:416)
at
org.apache.beam.runners.core.construction.graph.QueryablePipeline.buildNetwork(QueryablePipeline.java:176)
at
org.apache.beam.runners.core.construction.graph.QueryablePipeline.(QueryablePipeline.java:119)
at
org.apache.beam.runners.core.construction.graph.QueryablePipeline.forPrimitivesIn(QueryablePipeline.java:82)
at
org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.(GreedyPipelineFuser.java:67)
at
org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.fuse(GreedyPipelineFuser.java:89)
at
org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:96)
at
org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
at
org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
at
org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
{noformat}
Looks like it was caused by https://github.com/apache/beam/pull/6328
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5440) Add option to mount a directory inside SDK harness containers
[
https://issues.apache.org/jira/browse/BEAM-5440?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5440:
-
Description:
While experimenting with the Python SDK locally, I found it inconvenient that I
can't mount a host directory to the Docker containers, i.e. the input must
already be in the container and the results of a Write remain inside the
container. For local testing, users may want to mount a host directory.
Since BEAM-5288 the {{Environment}} carries explicit environment information,
we could a) add volume args to the {{DockerPayload}}, or b) provide a general
Docker arguments field.
was:
While experimenting with the Python SDK locally, I found it inconvenient that I
can't mount a host directory to the Docker containers, i.e. the input must
already be in the container and the results of a Write remain inside the
container. For local testing, users may want to mount a host directory.
Since the {{Environment}} now carries explicit environment information, we
could a) add volume args to the {{DockerPayload}}, or b) provide a general
Docker arguments field.
> Add option to mount a directory inside SDK harness containers
> -
>
> Key: BEAM-5440
> URL: https://issues.apache.org/jira/browse/BEAM-5440
> Project: Beam
> Issue Type: New Feature
> Components: java-fn-execution
>Reporter: Maximilian Michels
>Priority: Major
> Fix For: 2.8.0
>
>
> While experimenting with the Python SDK locally, I found it inconvenient that
> I can't mount a host directory to the Docker containers, i.e. the input must
> already be in the container and the results of a Write remain inside the
> container. For local testing, users may want to mount a host directory.
> Since BEAM-5288 the {{Environment}} carries explicit environment information,
> we could a) add volume args to the {{DockerPayload}}, or b) provide a general
> Docker arguments field.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5440) Add option to mount a directory inside SDK harness containers
[
https://issues.apache.org/jira/browse/BEAM-5440?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5440:
-
Description:
While experimenting with the Python SDK locally, I found it inconvenient that I
can't mount a host directory to the Docker containers, i.e. the input must
already be in the container and the results of a Write remain inside the
container. For local testing, users may want to mount a host directory.
Since the {{Environment}} now carries explicit environment information, we
could a) add volume args to the {{DockerPayload}}, or b) provide a general
Docker arguments field.
was:
While experimenting with the Python SDK locally, I found it inconvenient that I
can't mount a host directory to the Docker containers, i.e. the results of a
Write inside a container remain in the container. For local testing, users may
want to mount a host directory.
Since the {{Environment}} now carries explicit environment information, we
could a) add volume args to the {{DockerPayload}}, or b) provide a general
Docker arguments field.
> Add option to mount a directory inside SDK harness containers
> -
>
> Key: BEAM-5440
> URL: https://issues.apache.org/jira/browse/BEAM-5440
> Project: Beam
> Issue Type: New Feature
> Components: java-fn-execution
>Reporter: Maximilian Michels
>Priority: Major
> Fix For: 2.8.0
>
>
> While experimenting with the Python SDK locally, I found it inconvenient that
> I can't mount a host directory to the Docker containers, i.e. the input must
> already be in the container and the results of a Write remain inside the
> container. For local testing, users may want to mount a host directory.
> Since the {{Environment}} now carries explicit environment information, we
> could a) add volume args to the {{DockerPayload}}, or b) provide a general
> Docker arguments field.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5440) Add option to mount a directory inside SDK harness containers
Maximilian Michels created BEAM-5440:
Summary: Add option to mount a directory inside SDK harness
containers
Key: BEAM-5440
URL: https://issues.apache.org/jira/browse/BEAM-5440
Project: Beam
Issue Type: New Feature
Components: java-fn-execution
Reporter: Maximilian Michels
Fix For: 2.8.0
While experimenting with the Python SDK locally, I found it inconvenient that I
can't mount a host directory to the Docker containers, i.e. the results of a
Write inside a container remain in the container. For local testing, users may
want to mount a host directory.
Since the {{Environment}} now carries explicit environment information, we
could a) add volume args to the {{DockerPayload}}, or b) provide a general
Docker arguments field.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-3089) Issue with setting the parallelism at client level using Flink runner
[
https://issues.apache.org/jira/browse/BEAM-3089?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels resolved BEAM-3089.
--
Resolution: Fixed
Assignee: Maximilian Michels (was: Grzegorz Kołakowski)
Fixed via 38e3a8101a4c2269010fa2544be0acef5e740368
> Issue with setting the parallelism at client level using Flink runner
> -
>
> Key: BEAM-3089
> URL: https://issues.apache.org/jira/browse/BEAM-3089
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Affects Versions: 2.0.0
> Environment: I am using Flink 1.2.1 running on Docker, with Task
> Managers distributed across different VMs as part of a Docker Swarm.
>Reporter: Thalita Vergilio
>Assignee: Maximilian Michels
>Priority: Major
> Labels: docker, flink, parallel-deployment
> Fix For: 2.8.0
>
> Attachments: flink-ui-parallelism.png
>
> Time Spent: 3h
> Remaining Estimate: 0h
>
> When uploading an Apache Beam application using the Flink Web UI, the
> parallelism set at job submission doesn't get picked up. The same happens
> when submitting a job using the Flink CLI.
> In both cases, the parallelism ends up defaulting to 1.
> When I set the parallelism programmatically within the Apache Beam code, it
> works: {{flinkPipelineOptions.setParallelism(4);}}
> I suspect the root of the problem may be in the
> org.apache.beam.runners.flink.DefaultParallelismFactory class, as it checks
> for Flink's GlobalConfiguration, which may not pick up runtime values passed
> to Flink, then defaults to 1 if it doesn't find anything.
> Any ideas on how this could be fixed or worked around? I need to be able to
> change the parallelism dynamically, so the programmatic approach won't really
> work for me, nor will setting the Flink configuration at system level.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5418) Add Flink version compatibility table to Runner page
[ https://issues.apache.org/jira/browse/BEAM-5418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5418. -- Resolution: Fixed Fix Version/s: 2.8.0 > Add Flink version compatibility table to Runner page > - > > Key: BEAM-5418 > URL: https://issues.apache.org/jira/browse/BEAM-5418 > Project: Beam > Issue Type: Improvement > Components: runner-flink, website >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Major > Fix For: 2.8.0 > > Time Spent: 1h 10m > Remaining Estimate: 0h > > Following up a discussion on the mailing list. There have been confusions > which version of Beam is compatible with which Flink version. The only way > for users at the moment is to look into the source code. > A table like this will be helpful: > || Beam || Flink || > |2.5.0 | 1.4.0 | > |2.6.0 | 1.5.0 | > |2.7.0 | 1.5.2 | -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5385) Flink jobserver does not honor --flink-master-url
[ https://issues.apache.org/jira/browse/BEAM-5385?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5385: - Fix Version/s: (was: 2.8.0) 2.7.0 > Flink jobserver does not honor --flink-master-url > - > > Key: BEAM-5385 > URL: https://issues.apache.org/jira/browse/BEAM-5385 > Project: Beam > Issue Type: Bug > Components: runner-flink >Reporter: Thomas Weise >Assignee: Thomas Weise >Priority: Major > Fix For: 2.7.0 > > Time Spent: 40m > Remaining Estimate: 0h > > It will use the external Flink cluster when specified, but only with default > port number 8081, because the actual port is not in propagated in > FlinkExecutionEnvironments (RestOptions.PORT setting). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5239) Allow configure latencyTrackingInterval
[ https://issues.apache.org/jira/browse/BEAM-5239?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5239: - Fix Version/s: 2.7.0 > Allow configure latencyTrackingInterval > --- > > Key: BEAM-5239 > URL: https://issues.apache.org/jira/browse/BEAM-5239 > Project: Beam > Issue Type: New Feature > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.7.0 > > Time Spent: 3h 40m > Remaining Estimate: 0h > > Because of FLINK-10226, we need to be able to set > latencyTrackingConfiguration for flink via FlinkPipelineOptions -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5246) Beam metrics exported as flink metrics are not correct
[ https://issues.apache.org/jira/browse/BEAM-5246?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5246: - Fix Version/s: (was: 2.8.0) 2.7.0 > Beam metrics exported as flink metrics are not correct > -- > > Key: BEAM-5246 > URL: https://issues.apache.org/jira/browse/BEAM-5246 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.7.0 > > Time Spent: 1h 10m > Remaining Estimate: 0h > > In Flink UI and fink native MetricReported, I am seeing too many instances of > my Beam metric counter. It looks like the counter is materialised for every > operator running within the task, although is is emitter from only one beam > step (which should map to one operator?). This produces double counting. > A bit debugging I noticed this is happening for stream jobs. In batch I was > not able to reproduce it. Problem might be in FlinkMetricContainer. > [https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java#L86] > The update seems to be called from operators after finishing the bundle. Data > from accumulator are flushed to `runtimeContext.getMetricGroup()`. The scope > of accumulator seems to be different than metricGroup as in there with > different call the scope components change, especially for operatorID. It > seems like during the run, `metricResult.getStep()` does not match > operatorName of metricGroup where metric is being pushed. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5239) Allow configure latencyTrackingInterval
[ https://issues.apache.org/jira/browse/BEAM-5239?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5239: - Fix Version/s: (was: 2.8.0) > Allow configure latencyTrackingInterval > --- > > Key: BEAM-5239 > URL: https://issues.apache.org/jira/browse/BEAM-5239 > Project: Beam > Issue Type: New Feature > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.7.0 > > Time Spent: 3h 40m > Remaining Estimate: 0h > > Because of FLINK-10226, we need to be able to set > latencyTrackingConfiguration for flink via FlinkPipelineOptions -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-3089) Issue with setting the parallelism at client level using Flink runner
[
https://issues.apache.org/jira/browse/BEAM-3089?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-3089:
-
Fix Version/s: 2.8.0
> Issue with setting the parallelism at client level using Flink runner
> -
>
> Key: BEAM-3089
> URL: https://issues.apache.org/jira/browse/BEAM-3089
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Affects Versions: 2.0.0
> Environment: I am using Flink 1.2.1 running on Docker, with Task
> Managers distributed across different VMs as part of a Docker Swarm.
>Reporter: Thalita Vergilio
>Assignee: Grzegorz Kołakowski
>Priority: Major
> Labels: docker, flink, parallel-deployment
> Fix For: 2.8.0
>
> Attachments: flink-ui-parallelism.png
>
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> When uploading an Apache Beam application using the Flink Web UI, the
> parallelism set at job submission doesn't get picked up. The same happens
> when submitting a job using the Flink CLI.
> In both cases, the parallelism ends up defaulting to 1.
> When I set the parallelism programmatically within the Apache Beam code, it
> works: {{flinkPipelineOptions.setParallelism(4);}}
> I suspect the root of the problem may be in the
> org.apache.beam.runners.flink.DefaultParallelismFactory class, as it checks
> for Flink's GlobalConfiguration, which may not pick up runtime values passed
> to Flink, then defaults to 1 if it doesn't find anything.
> Any ideas on how this could be fixed or worked around? I need to be able to
> change the parallelism dynamically, so the programmatic approach won't really
> work for me, nor will setting the Flink configuration at system level.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5267) Update Flink Runner to Flink 1.6.x
[ https://issues.apache.org/jira/browse/BEAM-5267?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619100#comment-16619100 ] Maximilian Michels commented on BEAM-5267: -- We won't upgrade the main build target to 1.6 soon. Instead, an additional 1.6 target should be created. See BEAM-5419. > Update Flink Runner to Flink 1.6.x > -- > > Key: BEAM-5267 > URL: https://issues.apache.org/jira/browse/BEAM-5267 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Major > Fix For: 2.8.0 > > Time Spent: 40m > Remaining Estimate: 0h > > For the next release, the Flink version should be bumped. As changes for > 2.7.0 are already frozen, it's going to be 2.8.0. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-5419) Build multiple versions of the Flink Runner against different Flink versions
Maximilian Michels created BEAM-5419: Summary: Build multiple versions of the Flink Runner against different Flink versions Key: BEAM-5419 URL: https://issues.apache.org/jira/browse/BEAM-5419 Project: Beam Issue Type: New Feature Components: build-system, runner-flink Reporter: Maximilian Michels Assignee: Maximilian Michels Fix For: 2.8.0 Following up on a discussion on the mailing list. We want to keep the Flink version stable across different versions to avoid upgrade pain for long-term users. At the same time, there are users out there with newer Flink clusters and developers also want to utilize new Flink features. It would be great to build multiple versions of the Flink Runner against different Flink versions. When the upgrade is as simple as changing the version property in the build script, this should be pretty straight-forward. If not, having a "base version" and applying a patch during the build could be an option. We should avoid duplicating any Runner code. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-5418) Add Flink version compatibility table to Runner page
Maximilian Michels created BEAM-5418: Summary: Add Flink version compatibility table to Runner page Key: BEAM-5418 URL: https://issues.apache.org/jira/browse/BEAM-5418 Project: Beam Issue Type: Improvement Components: runner-flink, website Reporter: Maximilian Michels Assignee: Maximilian Michels Following up a discussion on the mailing list. There have been confusions which version of Beam is compatible with which Flink version. The only way for users at the moment is to look into the source code. A table like this will be helpful: || Beam || Flink || |2.5.0 | 1.4.0 | |2.6.0 | 1.5.0 | |2.7.0 | 1.5.2 | -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-593) Support unblocking run() in FlinkRunner and cancel() and waitUntilFinish() in FlinkRunnerResult
[ https://issues.apache.org/jira/browse/BEAM-593?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16618916#comment-16618916 ] Maximilian Michels commented on BEAM-593: - Unfortunately, this is not resolved for the non-portable FlinkRunner. However, it works for the portable Runner. > Support unblocking run() in FlinkRunner and cancel() and waitUntilFinish() in > FlinkRunnerResult > --- > > Key: BEAM-593 > URL: https://issues.apache.org/jira/browse/BEAM-593 > Project: Beam > Issue Type: New Feature > Components: runner-flink >Reporter: Pei He >Assignee: Aljoscha Krettek >Priority: Major > Time Spent: 1h > Remaining Estimate: 0h > > We introduced both functions to PipelineResult. > Currently, both of them throw UnsupportedOperationException in Flink runner. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Closed] (BEAM-5405) Remove deprecated AbstractStateBackend from FlinkPipelineOptions
[
https://issues.apache.org/jira/browse/BEAM-5405?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels closed BEAM-5405.
Resolution: Fixed
23eb70ffa5b96396f75391364d1ff37f9ce6604a
> Remove deprecated AbstractStateBackend from FlinkPipelineOptions
>
>
> Key: BEAM-5405
> URL: https://issues.apache.org/jira/browse/BEAM-5405
> Project: Beam
> Issue Type: Improvement
> Components: runner-flink
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Minor
> Fix For: 2.8.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> There is now an interface called {{StateBackend}} which is the superclass of
> {{AbstractStateBackend}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5396) Flink portable runner savepoint / upgrade support
[ https://issues.apache.org/jira/browse/BEAM-5396?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16617869#comment-16617869 ] Maximilian Michels commented on BEAM-5396: -- This is not an easy one. Once the portability is stable, we can start assigning UUIDs to all the operators and carefully migrate state whenever necessary. Additionally, we will have to have regression tests to ensure that Beam jobs can be upgraded from one version to another. Some background: https://ci.apache.org/projects/flink/flink-docs-stable/ops/upgrading.html > Flink portable runner savepoint / upgrade support > - > > Key: BEAM-5396 > URL: https://issues.apache.org/jira/browse/BEAM-5396 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Major > > The Flink runner overall and the new portable implementation specifically > need to support Flink savepoints for production use. Specifically, it should > be possible to upgrade a stateful portable Beam pipeline that runs on Flink, > which involves taking a savepoint and then starting the new version of the > pipeline from that savepoint. The potential issues with pipeline evolution > and migration are similar to those when using the Flink DataStream API > (schema / name changes etc.). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Closed] (BEAM-5403) Update Flink Runner to 1.5.3
[ https://issues.apache.org/jira/browse/BEAM-5403?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels closed BEAM-5403. Resolution: Fixed 79fb76b63309426820ccd0ee874c45d052db3b70 > Update Flink Runner to 1.5.3 > > > Key: BEAM-5403 > URL: https://issues.apache.org/jira/browse/BEAM-5403 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Maximilian Michels >Assignee: Maximilian Michels >Priority: Minor > Fix For: 2.8.0 > > Time Spent: 40m > Remaining Estimate: 0h > > We should be on the latest hotfix version to ensure stability for users. > https://flink.apache.org/news/2018/08/21/release-1.5.3.html > The only fix affecting Beam users seems to be FLINK-9289. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-5405) Remove deprecated AbstractStateBackend from FlinkPipelineOptions
Maximilian Michels created BEAM-5405:
Summary: Remove deprecated AbstractStateBackend from
FlinkPipelineOptions
Key: BEAM-5405
URL: https://issues.apache.org/jira/browse/BEAM-5405
Project: Beam
Issue Type: Improvement
Components: runner-flink
Reporter: Maximilian Michels
Assignee: Maximilian Michels
Fix For: 2.8.0
There is now an interface called {{StateBackend}} which is the superclass of
{{AbstractStateBackend}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-5403) Update Flink Runner to 1.5.3
Maximilian Michels created BEAM-5403: Summary: Update Flink Runner to 1.5.3 Key: BEAM-5403 URL: https://issues.apache.org/jira/browse/BEAM-5403 Project: Beam Issue Type: Improvement Components: runner-flink Reporter: Maximilian Michels Assignee: Maximilian Michels Fix For: 2.8.0 We should be on the latest hotfix version to ensure stability for users. https://flink.apache.org/news/2018/08/21/release-1.5.3.html The only fix affecting Beam users seems to be FLINK-9289. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5337) [beam_PostCommit_Java_GradleBuild][:beam-runners-flink_2.11:test][Flake] Build times out in beam-runners-flink target
[ https://issues.apache.org/jira/browse/BEAM-5337?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5337. -- Resolution: Fixed Fix Version/s: 2.8.0 This has been resolved with faa5699e79193b17c0255cd318892c5e0db8a4de. Please reopen if you see any more failures. > [beam_PostCommit_Java_GradleBuild][:beam-runners-flink_2.11:test][Flake] > Build times out in beam-runners-flink target > - > > Key: BEAM-5337 > URL: https://issues.apache.org/jira/browse/BEAM-5337 > Project: Beam > Issue Type: Bug > Components: runner-flink, test-failures >Reporter: Mikhail Gryzykhin >Assignee: Maximilian Michels >Priority: Critical > Fix For: 2.8.0 > > Time Spent: 2h > Remaining Estimate: 0h > > Job times out. > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/1414/consoleFull] > [https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/1406/consoleFull] > https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/1408/consoleFull > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5246) Beam metrics exported as flink metrics are not correct
[ https://issues.apache.org/jira/browse/BEAM-5246?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5246. -- Resolution: Fixed Fix Version/s: (was: 2.7.0) 2.8.0 Should be resolved now. Thanks [~JozoVilcek]. > Beam metrics exported as flink metrics are not correct > -- > > Key: BEAM-5246 > URL: https://issues.apache.org/jira/browse/BEAM-5246 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.8.0 > > Time Spent: 1h 10m > Remaining Estimate: 0h > > In Flink UI and fink native MetricReported, I am seeing too many instances of > my Beam metric counter. It looks like the counter is materialised for every > operator running within the task, although is is emitter from only one beam > step (which should map to one operator?). This produces double counting. > A bit debugging I noticed this is happening for stream jobs. In batch I was > not able to reproduce it. Problem might be in FlinkMetricContainer. > [https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java#L86] > The update seems to be called from operators after finishing the bundle. Data > from accumulator are flushed to `runtimeContext.getMetricGroup()`. The scope > of accumulator seems to be different than metricGroup as in there with > different call the scope components change, especially for operatorID. It > seems like during the run, `metricResult.getStep()` does not match > operatorName of metricGroup where metric is being pushed. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5337) [beam_PostCommit_Java_GradleBuild][:beam-runners-flink_2.11:test][Flake] Build times out in beam-runners-flink target
[ https://issues.apache.org/jira/browse/BEAM-5337?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5337: - Component/s: runner-flink > [beam_PostCommit_Java_GradleBuild][:beam-runners-flink_2.11:test][Flake] > Build times out in beam-runners-flink target > - > > Key: BEAM-5337 > URL: https://issues.apache.org/jira/browse/BEAM-5337 > Project: Beam > Issue Type: Bug > Components: runner-flink, test-failures >Reporter: Mikhail Gryzykhin >Assignee: Aljoscha Krettek >Priority: Major > > Job times out. > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/1414/consoleFull] > [https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/1406/consoleFull] > https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/1408/consoleFull > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (BEAM-5288) Modify Environment to support non-dockerized SDK harness deployments
[
https://issues.apache.org/jira/browse/BEAM-5288?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels reassigned BEAM-5288:
Assignee: Ankur Goenka (was: Maximilian Michels)
> Modify Environment to support non-dockerized SDK harness deployments
> -
>
> Key: BEAM-5288
> URL: https://issues.apache.org/jira/browse/BEAM-5288
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
>Reporter: Maximilian Michels
>Assignee: Ankur Goenka
>Priority: Major
> Time Spent: 40m
> Remaining Estimate: 0h
>
> As of mailing discussions and BEAM-5187, it has become clear that we need to
> extend the Environment information. In addition to the Docker environment,
> the extended environment holds deployment options for 1) a process-based
> environment, 2) an externally managed environment.
> The proto definition, as of now, looks as follows:
> {noformat}
> message Environment {
>// (Required) The URN of the payload
>string urn = 1;
>// (Optional) The data specifying any parameters to the URN. If
>// the URN does not require any arguments, this may be omitted.
>bytes payload = 2;
> }
> message StandardEnvironments {
>enum Environments {
> DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
> PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
> EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
>}
> }
> // The payload of a Docker image
> message DockerPayload {
>string container_image = 1; // implicitly linux_amd64.
> }
> message ProcessPayload {
>string os = 1; // "linux", "darwin", ..
>string arch = 2; // "amd64", ..
>string command = 3; // process to execute
>map env = 4; // environment variables
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5299) Define max global window as a shared value in protos like URN enums.
[
https://issues.apache.org/jira/browse/BEAM-5299?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16611025#comment-16611025
]
Maximilian Michels commented on BEAM-5299:
--
Thanks for the pointer [~lcwik]. What I did was adding another EnumValueOption:
{code:java}
extend google.protobuf.EnumValueOptions {
string beam_urn = 185324356;
int64 global_window_max_timestamp_millis = 185324357;
}
{code}
But it makes much more sense to have a string constant field. Or even to reuse
beam_urn and make it generic:
{code:java}
extend google.protobuf.EnumValueOptions {
string beam_constant = 185324356;
}
message Constants {
enum Constants {
// The minimum timestamp in milliseconds since Jan 1, 1970
TIMESTAMP_MIN_MILLIS = 0 [(beam_constant) = "-9223372036854775"];
// The maximum timestamp in milliseconds since Jan 1, 1970
TIMESTAMP_MAX_MILLIS = 0 [(beam_constant) = "9223372036854775"];
// The maximum timestamp for the global window in milliseconds since Jan 1,
1970
// Triggers use maxTimestamp to set timers' timestamp. Timers fires when
// the watermark passes their timestamps. So, the timestamp needs to be
// smaller than the TIMESTAMP_MAX_MILLIS.
GLOBAL_WINDOW_MAX_TIMESTAMP_MILLIS = 1 [(beam_constant) =
"9223371950454775"];
}
}
{code}
> Define max global window as a shared value in protos like URN enums.
>
>
> Key: BEAM-5299
> URL: https://issues.apache.org/jira/browse/BEAM-5299
> Project: Beam
> Issue Type: Improvement
> Components: beam-model, sdk-go, sdk-java-core, sdk-py-core
>Reporter: Luke Cwik
>Assignee: Maximilian Michels
>Priority: Minor
> Labels: portability
>
> Instead of having each language define a max timestamp themselves, define the
> max timestamps within proto to be shared across different languages.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (BEAM-4615) Flink job server driver wrapper
[
https://issues.apache.org/jira/browse/BEAM-4615?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16610895#comment-16610895
]
Maximilian Michels edited comment on BEAM-4615 at 9/11/18 4:29 PM:
---
bq. A gradle wrapper that can execute the Flink job server driver so that it
can be easily run locally for testing/debugging.
The first part of this is done.
bq. A shadow ("uber") target that packages all portable Flink runner
dependencies into a runnable jar. This jar can then be submitted to Flink
clusters via `flink run`.
[~bsidhom] What exactly is the second task? Should this be a bundled pipeline
which uses a local JobServer which is spawned during job execution?
was (Author: mxm):
bq. A gradle wrapper that can execute the Flink job server driver so that it
can be easily run locally for testing/debugging.
The first part of this is done.
bq. A shadow ("uber") target that packages all portable Flink runner
dependencies into a runnable jar. This jar can then be submitted to Flink
clusters via `flink run`.
[~bsidhom] What exactly is the second task? Should this be a bundled pipeline
which uses a local JobServer?
> Flink job server driver wrapper
> ---
>
> Key: BEAM-4615
> URL: https://issues.apache.org/jira/browse/BEAM-4615
> Project: Beam
> Issue Type: New Feature
> Components: runner-flink
>Reporter: Ben Sidhom
>Assignee: Ben Sidhom
>Priority: Major
> Time Spent: 5h 50m
> Remaining Estimate: 0h
>
> This includes:
> * A gradle wrapper that can execute the Flink job server driver so that it
> can be easily run locally for testing/debugging.
> * A shadow ("uber") target that packages all portable Flink runner
> dependencies into a runnable jar. This jar can then be submitted to Flink
> clusters via `flink run`.
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-4615) Flink job server driver wrapper
[
https://issues.apache.org/jira/browse/BEAM-4615?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16610895#comment-16610895
]
Maximilian Michels commented on BEAM-4615:
--
bq. A gradle wrapper that can execute the Flink job server driver so that it
can be easily run locally for testing/debugging.
The first part of this is done.
bq. A shadow ("uber") target that packages all portable Flink runner
dependencies into a runnable jar. This jar can then be submitted to Flink
clusters via `flink run`.
[~bsidhom] What exactly is the second task? Should this be a bundled pipeline
which uses a local JobServer?
> Flink job server driver wrapper
> ---
>
> Key: BEAM-4615
> URL: https://issues.apache.org/jira/browse/BEAM-4615
> Project: Beam
> Issue Type: New Feature
> Components: runner-flink
>Reporter: Ben Sidhom
>Assignee: Ben Sidhom
>Priority: Major
> Time Spent: 5h 50m
> Remaining Estimate: 0h
>
> This includes:
> * A gradle wrapper that can execute the Flink job server driver so that it
> can be easily run locally for testing/debugging.
> * A shadow ("uber") target that packages all portable Flink runner
> dependencies into a runnable jar. This jar can then be submitted to Flink
> clusters via `flink run`.
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5239) Allow configure latencyTrackingInterval
[ https://issues.apache.org/jira/browse/BEAM-5239?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5239. -- Resolution: Fixed Fix Version/s: (was: 2.7.0) 2.8.0 > Allow configure latencyTrackingInterval > --- > > Key: BEAM-5239 > URL: https://issues.apache.org/jira/browse/BEAM-5239 > Project: Beam > Issue Type: New Feature > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.8.0 > > Time Spent: 3h 20m > Remaining Estimate: 0h > > Because of FLINK-10226, we need to be able to set > latencyTrackingConfiguration for flink via FlinkPipelineOptions -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5246) Beam metrics exported as flink metrics are not correct
[ https://issues.apache.org/jira/browse/BEAM-5246?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5246: - Fix Version/s: 2.7.0 > Beam metrics exported as flink metrics are not correct > -- > > Key: BEAM-5246 > URL: https://issues.apache.org/jira/browse/BEAM-5246 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.7.0 > > Time Spent: 20m > Remaining Estimate: 0h > > In Flink UI and fink native MetricReported, I am seeing too many instances of > my Beam metric counter. It looks like the counter is materialised for every > operator running within the task, although is is emitter from only one beam > step (which should map to one operator?). This produces double counting. > A bit debugging I noticed this is happening for stream jobs. In batch I was > not able to reproduce it. Problem might be in FlinkMetricContainer. > [https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java#L86] > The update seems to be called from operators after finishing the bundle. Data > from accumulator are flushed to `runtimeContext.getMetricGroup()`. The scope > of accumulator seems to be different than metricGroup as in there with > different call the scope components change, especially for operatorID. It > seems like during the run, `metricResult.getStep()` does not match > operatorName of metricGroup where metric is being pushed. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (BEAM-5246) Beam metrics exported as flink metrics are not correct
[ https://issues.apache.org/jira/browse/BEAM-5246?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels reassigned BEAM-5246: Assignee: Jozef Vilcek (was: Aljoscha Krettek) > Beam metrics exported as flink metrics are not correct > -- > > Key: BEAM-5246 > URL: https://issues.apache.org/jira/browse/BEAM-5246 > Project: Beam > Issue Type: Bug > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.7.0 > > Time Spent: 20m > Remaining Estimate: 0h > > In Flink UI and fink native MetricReported, I am seeing too many instances of > my Beam metric counter. It looks like the counter is materialised for every > operator running within the task, although is is emitter from only one beam > step (which should map to one operator?). This produces double counting. > A bit debugging I noticed this is happening for stream jobs. In batch I was > not able to reproduce it. Problem might be in FlinkMetricContainer. > [https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java#L86] > The update seems to be called from operators after finishing the bundle. Data > from accumulator are flushed to `runtimeContext.getMetricGroup()`. The scope > of accumulator seems to be different than metricGroup as in there with > different call the scope components change, especially for operatorID. It > seems like during the run, `metricResult.getStep()` does not match > operatorName of metricGroup where metric is being pushed. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5239) Allow configure latencyTrackingInterval
[ https://issues.apache.org/jira/browse/BEAM-5239?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels updated BEAM-5239: - Fix Version/s: 2.7.0 > Allow configure latencyTrackingInterval > --- > > Key: BEAM-5239 > URL: https://issues.apache.org/jira/browse/BEAM-5239 > Project: Beam > Issue Type: New Feature > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Fix For: 2.7.0 > > Time Spent: 3h > Remaining Estimate: 0h > > Because of FLINK-10226, we need to be able to set > latencyTrackingConfiguration for flink via FlinkPipelineOptions -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (BEAM-5239) Allow configure latencyTrackingInterval
[ https://issues.apache.org/jira/browse/BEAM-5239?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels reassigned BEAM-5239: Assignee: Jozef Vilcek (was: Aljoscha Krettek) > Allow configure latencyTrackingInterval > --- > > Key: BEAM-5239 > URL: https://issues.apache.org/jira/browse/BEAM-5239 > Project: Beam > Issue Type: New Feature > Components: runner-flink >Affects Versions: 2.6.0 >Reporter: Jozef Vilcek >Assignee: Jozef Vilcek >Priority: Major > Time Spent: 3h > Remaining Estimate: 0h > > Because of FLINK-10226, we need to be able to set > latencyTrackingConfiguration for flink via FlinkPipelineOptions -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5299) Define max global window as a shared value in protos like URN enums.
[
https://issues.apache.org/jira/browse/BEAM-5299?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16610725#comment-16610725
]
Maximilian Michels commented on BEAM-5299:
--
That is exactly the type of lookup I had in mind. I had added the timestamp to
the enum value extension but this felt rather hacky. The {{beam_urn}} extension
is applicable to every transform, the max timestamp value only applies to
PAR_DO and ASSIGN_WINDOW, but ideally it should be a global constant.
I'm not sure what the best point is to encode a proto-wide constant.
> Define max global window as a shared value in protos like URN enums.
>
>
> Key: BEAM-5299
> URL: https://issues.apache.org/jira/browse/BEAM-5299
> Project: Beam
> Issue Type: Improvement
> Components: beam-model, sdk-go, sdk-java-core, sdk-py-core
>Reporter: Luke Cwik
>Assignee: Maximilian Michels
>Priority: Minor
> Labels: portability
>
> Instead of having each language define a max timestamp themselves, define the
> max timestamps within proto to be shared across different languages.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5299) Define max global window as a shared value in protos like URN enums.
[
https://issues.apache.org/jira/browse/BEAM-5299?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16609599#comment-16609599
]
Maximilian Michels commented on BEAM-5299:
--
DoFns are free to return WindowedValues with GlobalWindow, but they don't have
a notion of the Proto (e.g. at least in the JAVA SDK when you look at
{{WindowedValue}}). So instead of using a concrete timestamp for the global
window, we would have to use a placeholder which gets translated into a lookup
at runtime.
I'm leaning more towards documenting and fixing the global max timestamp value
in all SDKs which doesn't seem that hard.
Any thoughts on this [~lcwik]?
> Define max global window as a shared value in protos like URN enums.
>
>
> Key: BEAM-5299
> URL: https://issues.apache.org/jira/browse/BEAM-5299
> Project: Beam
> Issue Type: Improvement
> Components: beam-model, sdk-go, sdk-java-core, sdk-py-core
>Reporter: Luke Cwik
>Assignee: Maximilian Michels
>Priority: Minor
> Labels: portability
>
> Instead of having each language define a max timestamp themselves, define the
> max timestamps within proto to be shared across different languages.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (BEAM-5288) Modify Environment to support non-dockerized SDK harness deployments
[
https://issues.apache.org/jira/browse/BEAM-5288?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels reassigned BEAM-5288:
Assignee: Maximilian Michels
> Modify Environment to support non-dockerized SDK harness deployments
> -
>
> Key: BEAM-5288
> URL: https://issues.apache.org/jira/browse/BEAM-5288
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
>
> As of mailing discussions and BEAM-5187, it has become clear that we need to
> extend the Environment information. In addition to the Docker environment,
> the extended environment holds deployment options for 1) a process-based
> environment, 2) an externally managed environment.
> The proto definition, as of now, looks as follows:
> {noformat}
> message Environment {
>// (Required) The URN of the payload
>string urn = 1;
>// (Optional) The data specifying any parameters to the URN. If
>// the URN does not require any arguments, this may be omitted.
>bytes payload = 2;
> }
> message StandardEnvironments {
>enum Environments {
> DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
> PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
> EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
>}
> }
> // The payload of a Docker image
> message DockerPayload {
>string container_image = 1; // implicitly linux_amd64.
> }
> message ProcessPayload {
>string os = 1; // "linux", "darwin", ..
>string arch = 2; // "amd64", ..
>string command = 3; // process to execute
>map env = 4; // environment variables
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5332) SDK harness containers are not eventually shut down after job ends
[
https://issues.apache.org/jira/browse/BEAM-5332?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels resolved BEAM-5332.
--
Resolution: Fixed
Fixed with https://github.com/apache/beam/pull/6342
> SDK harness containers are not eventually shut down after job ends
> --
>
> Key: BEAM-5332
> URL: https://issues.apache.org/jira/browse/BEAM-5332
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>Priority: Major
> Fix For: 2.8.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> When the job shuts down, the user code classloader is cleared which removes
> the possibility to load new classes. The {{LoadingCache}} attempts to load
> the {{RemovalCause}} class after job shutdown to evict the cache.
> We shouldn't attempt to execute code after the job has been removed. This is
> not safe, at least not with Flink.
> {noformat}
> 2018-09-06 15:37:07,996 ERROR
> org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory
> - Unable to close.
> java.lang.NoClassDefFoundError:
> org/apache/beam/repackaged/beam_runners_java_fn_execution/com/google/common/cache/RemovalCause
> at
> org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache$Segment.clear(LocalCache.java:3290)
> at
> org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache.clear(LocalCache.java:4322)
> at
> org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache$LocalManualCache.invalidateAll(LocalCache.java:4937)
> at
> org.apache.beam.runners.fnexecution.control.JobBundleFactoryBase.close(JobBundleFactoryBase.java:186)
> at
> org.apache.beam.runners.flink.translation.functions.FlinkBatchExecutableStageContext.close(FlinkBatchExecutableStageContext.java:68)
> at
> org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory$WrappedContext.closeActual(ReferenceCountingFlinkExecutableStageContextFactory.java:186)
> at
> org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory$WrappedContext.access$200(ReferenceCountingFlinkExecutableStageContextFactory.java:162)
> at
> org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory.release(ReferenceCountingFlinkExecutableStageContextFactory.java:150)
> at
> org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory.lambda$scheduleRelease$1(ReferenceCountingFlinkExecutableStageContextFactory.java:110)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
> at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.ClassNotFoundException:
> org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.RemovalCause
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at
> org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$ChildFirstClassLoader.loadClass(FlinkUserCodeClassLoaders.java:129)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> ... 16 more
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-5331) Flink portable runner Python validate runner failures
[ https://issues.apache.org/jira/browse/BEAM-5331?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5331. -- Resolution: Fixed Fix Version/s: 2.8.0 This is now resolved with the recent changes. > Flink portable runner Python validate runner failures > -- > > Key: BEAM-5331 > URL: https://issues.apache.org/jira/browse/BEAM-5331 > Project: Beam > Issue Type: Task > Components: runner-flink >Reporter: Thomas Weise >Priority: Major > Labels: portability > Fix For: 2.8.0 > > > Currently 2 Python validate runner tests fail on Flink (in streaming mode). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-5338) PortableValidatesRunner always rebuilds Python SDK Harness image
Maximilian Michels created BEAM-5338: Summary: PortableValidatesRunner always rebuilds Python SDK Harness image Key: BEAM-5338 URL: https://issues.apache.org/jira/browse/BEAM-5338 Project: Beam Issue Type: Improvement Components: build-system Reporter: Maximilian Michels Assignee: Luke Cwik The portable ValidatesRunner tests use the Python SDK Harness and requires a Docker image to be built. Building the image takes a long time (~20 minutes) and it would be nice if the build system only rebuilt if changes were made to the SDK harness. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5308) JobBundleFactory BindException with FlinkRunner and remote cluster
[ https://issues.apache.org/jira/browse/BEAM-5308?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5308. -- Resolution: Fixed Fix Version/s: 2.8.0 The root cause of this has been fixed, new unrelated followup: BEAM-5332 > JobBundleFactory BindException with FlinkRunner and remote cluster > -- > > Key: BEAM-5308 > URL: https://issues.apache.org/jira/browse/BEAM-5308 > Project: Beam > Issue Type: Task > Components: runner-flink >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability > Fix For: 2.8.0 > > Time Spent: 0.5h > Remaining Estimate: 0h > > Repeated execution of the same job on remote Flink cluster (not embedded in > job server) fails with bind exception. There seem to be 2 issues: > * Multiple instances of job bundle factory cannot be created (port conflict) > * Job bundle factory is not released after job completes (and Docker > container keeps on running). That's not the case in embedded mode). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-5332) SDK harness containers are not eventually shut down after job ends
Maximilian Michels created BEAM-5332:
Summary: SDK harness containers are not eventually shut down after
job ends
Key: BEAM-5332
URL: https://issues.apache.org/jira/browse/BEAM-5332
Project: Beam
Issue Type: Bug
Components: runner-flink
Reporter: Maximilian Michels
Assignee: Maximilian Michels
Fix For: 2.8.0
When the job shuts down, the user code classloader is cleared which removes the
possibility to load new classes. The {{LoadingCache}} attempts to load the
{{RemovalCause}} class after job shutdown to evict the cache.
We shouldn't attempt to execute code after the job has been removed. This is
not safe, at least not with Flink.
{noformat}
2018-09-06 15:37:07,996 ERROR
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory
- Unable to close.
java.lang.NoClassDefFoundError:
org/apache/beam/repackaged/beam_runners_java_fn_execution/com/google/common/cache/RemovalCause
at
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache$Segment.clear(LocalCache.java:3290)
at
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache.clear(LocalCache.java:4322)
at
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache$LocalManualCache.invalidateAll(LocalCache.java:4937)
at
org.apache.beam.runners.fnexecution.control.JobBundleFactoryBase.close(JobBundleFactoryBase.java:186)
at
org.apache.beam.runners.flink.translation.functions.FlinkBatchExecutableStageContext.close(FlinkBatchExecutableStageContext.java:68)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory$WrappedContext.closeActual(ReferenceCountingFlinkExecutableStageContextFactory.java:186)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory$WrappedContext.access$200(ReferenceCountingFlinkExecutableStageContextFactory.java:162)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory.release(ReferenceCountingFlinkExecutableStageContextFactory.java:150)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory.lambda$scheduleRelease$1(ReferenceCountingFlinkExecutableStageContextFactory.java:110)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException:
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.RemovalCause
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at
org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$ChildFirstClassLoader.loadClass(FlinkUserCodeClassLoaders.java:129)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 16 more
{noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5308) JobBundleFactory BindException with FlinkRunner and remote cluster
[
https://issues.apache.org/jira/browse/BEAM-5308?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16605791#comment-16605791
]
Maximilian Michels commented on BEAM-5308:
--
This is a classloading issue when closing the environment LoadingCache. The
exception was swallowed:
{noformat}
2018-09-06 15:37:07,996 ERROR
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory
- Unable to close.
java.lang.NoClassDefFoundError:
org/apache/beam/repackaged/beam_runners_java_fn_execution/com/google/common/cache/RemovalCause
at
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache$Segment.clear(LocalCache.java:3290)
at
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache.clear(LocalCache.java:4322)
at
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.LocalCache$LocalManualCache.invalidateAll(LocalCache.java:4937)
at
org.apache.beam.runners.fnexecution.control.JobBundleFactoryBase.close(JobBundleFactoryBase.java:186)
at
org.apache.beam.runners.flink.translation.functions.FlinkBatchExecutableStageContext.close(FlinkBatchExecutableStageContext.java:68)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory$WrappedContext.closeActual(ReferenceCountingFlinkExecutableStageContextFactory.java:186)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory$WrappedContext.access$200(ReferenceCountingFlinkExecutableStageContextFactory.java:162)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory.release(ReferenceCountingFlinkExecutableStageContextFactory.java:150)
at
org.apache.beam.runners.flink.translation.functions.ReferenceCountingFlinkExecutableStageContextFactory.lambda$scheduleRelease$1(ReferenceCountingFlinkExecutableStageContextFactory.java:110)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException:
org.apache.beam.repackaged.beam_runners_java_fn_execution.com.google.common.cache.RemovalCause
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at
org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$ChildFirstClassLoader.loadClass(FlinkUserCodeClassLoaders.java:129)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 16 more
{noformat}
> JobBundleFactory BindException with FlinkRunner and remote cluster
> --
>
> Key: BEAM-5308
> URL: https://issues.apache.org/jira/browse/BEAM-5308
> Project: Beam
> Issue Type: Task
> Components: runner-flink
>Reporter: Thomas Weise
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> Repeated execution of the same job on remote Flink cluster (not embedded in
> job server) fails with bind exception. There seem to be 2 issues:
> * Multiple instances of job bundle factory cannot be created (port conflict)
> * Job bundle factory is not released after job completes (and Docker
> container keeps on running). That's not the case in embedded mode).
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (BEAM-5288) Modify Environment to support non-dockerized SDK harness deployments
[
https://issues.apache.org/jira/browse/BEAM-5288?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels updated BEAM-5288:
-
Description:
As of mailing discussions and BEAM-5187, it has become clear that we need to
extend the Environment information. In addition to the Docker environment, the
extended environment holds deployment options for 1) a process-based
environment, 2) an externally managed environment.
The proto definition, as of now, looks as follows:
{noformat}
message Environment {
// (Required) The URN of the payload
string urn = 1;
// (Optional) The data specifying any parameters to the URN. If
// the URN does not require any arguments, this may be omitted.
bytes payload = 2;
}
message StandardEnvironments {
enum Environments {
DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
}
}
// The payload of a Docker image
message DockerPayload {
string container_image = 1; // implicitly linux_amd64.
}
message ProcessPayload {
string os = 1; // "linux", "darwin", ..
string arch = 2; // "amd64", ..
string command = 3; // process to execute
map env = 4; // environment variables
}
{noformat}
was:
As of mailing discussions and BEAM-5187, it has become clear that we need to
extend the Environment information. In addition to the Docker environment, the
extended environment holds deployment options for 1) a process-based
environment, 2) an externally managed environment.
The proto definition, as of now, looks as follows:
{noformat}
message Environment {
// (Required) The URN of the payload
string urn = 1;
// (Optional) The data specifying any parameters to the URN. If
// the URN does not require any arguments, this may be omitted.
bytes payload = 2;
}
message StandardEnvironments {
enum Environments {
DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
}
}
// The payload of a Docker image
message DockerPayload {
string container_image = 1; // implicitly linux_amd64.
}
message ProcessPayload {
string os = 1; // "linux", "darwin", ..
string arch = 2; // "amd64", ..
string command = 3; // process to execute
repeated string params = 4; // parameters
}
{noformat}
> Modify Environment to support non-dockerized SDK harness deployments
> -
>
> Key: BEAM-5288
> URL: https://issues.apache.org/jira/browse/BEAM-5288
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
>Reporter: Maximilian Michels
>Priority: Major
>
> As of mailing discussions and BEAM-5187, it has become clear that we need to
> extend the Environment information. In addition to the Docker environment,
> the extended environment holds deployment options for 1) a process-based
> environment, 2) an externally managed environment.
> The proto definition, as of now, looks as follows:
> {noformat}
> message Environment {
>// (Required) The URN of the payload
>string urn = 1;
>// (Optional) The data specifying any parameters to the URN. If
>// the URN does not require any arguments, this may be omitted.
>bytes payload = 2;
> }
> message StandardEnvironments {
>enum Environments {
> DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
> PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
> EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
>}
> }
> // The payload of a Docker image
> message DockerPayload {
>string container_image = 1; // implicitly linux_amd64.
> }
> message ProcessPayload {
>string os = 1; // "linux", "darwin", ..
>string arch = 2; // "amd64", ..
>string command = 3; // process to execute
>map env = 4; // environment variables
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5288) Modify Environment to support non-dockerized SDK harness deployments
[
https://issues.apache.org/jira/browse/BEAM-5288?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16605603#comment-16605603
]
Maximilian Michels commented on BEAM-5288:
--
[~lcwik] +1 {{SupportedEnvironments}} is better. Thanks for the adjustments.
[~herohde] Params would be additional arguments after the container contract
parameters. IMHO can be removed since the semantics are not clear, e.g. doesn't
work if the process expects positional parameters. We could add environment
variables instead as a way to pass parameters to the process.
> Modify Environment to support non-dockerized SDK harness deployments
> -
>
> Key: BEAM-5288
> URL: https://issues.apache.org/jira/browse/BEAM-5288
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
>Reporter: Maximilian Michels
>Priority: Major
>
> As of mailing discussions and BEAM-5187, it has become clear that we need to
> extend the Environment information. In addition to the Docker environment,
> the extended environment holds deployment options for 1) a process-based
> environment, 2) an externally managed environment.
> The proto definition, as of now, looks as follows:
> {noformat}
> message Environment {
>// (Required) The URN of the payload
>string urn = 1;
>// (Optional) The data specifying any parameters to the URN. If
>// the URN does not require any arguments, this may be omitted.
>bytes payload = 2;
> }
> message StandardEnvironments {
>enum Environments {
> DOCKER = 0 [(beam_urn) = "beam:env:docker:v1"];
> PROCESS = 1 [(beam_urn) = "beam:env:process:v1"];
> EXTERNAL = 2 [(beam_urn) = "beam:env:external:v1"];
>}
> }
> // The payload of a Docker image
> message DockerPayload {
>string container_image = 1; // implicitly linux_amd64.
> }
> message ProcessPayload {
>string os = 1; // "linux", "darwin", ..
>string arch = 2; // "amd64", ..
>string command = 3; // process to execute
>repeated string params = 4; // parameters
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (BEAM-5308) JobBundleFactory BindException with FlinkRunner and remote cluster
[ https://issues.apache.org/jira/browse/BEAM-5308?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels reassigned BEAM-5308: Assignee: Maximilian Michels > JobBundleFactory BindException with FlinkRunner and remote cluster > -- > > Key: BEAM-5308 > URL: https://issues.apache.org/jira/browse/BEAM-5308 > Project: Beam > Issue Type: Task > Components: runner-flink >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability > > Repeated execution of the same job on remote Flink cluster (not embedded in > job server) fails with bind exception. There seem to be 2 issues: > * Multiple instances of job bundle factory cannot be created (port conflict) > * Job bundle factory is not released after job completes (and Docker > container keeps on running). That's not the case in embedded mode). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
