Re: Beam 2.19.0 / Flink 1.9.1 - Session cluster error when submitting job "Multiple environments cannot be created in detached mode"

Maximilian Michels Mon, 02 Mar 2020 14:37:17 -0800

Hi Tobias,

You are right, for this use case where Kafka commits the offset and youdo not use any other stateful sources/operators you will get exactly-once.

Should you have stateful operators that may not hold true anymore. Onlya checkpoint constructs a consistent view of the pipeline. Whether youuse stop or cancel after taking the checkpoint does not make adifference in terms of correctness. To get exactly-once afterrestarting, you have to rewind to the checkpoint, or replay all datafrom the beginning.


But I understand that stop makes sense for the use cases you described.

-Max

On 02.03.20 16:29, Kaymak, Tobias wrote:

Hi Max,

thank you I've created the JIRA issue as requested:https://issues.apache.org/jira/browse/BEAM-9423

What I am struggling to understand is the exactly-once problem. When apipeline gets stopped, cleanly shuts down and there are no interruptions(taskmanager / jobmanager are all fine and do not get killed) - and thenit gets restarted and takes the last offset committed to Kafka - Iwouldn't expect any data loss or duplication. Is there something I ammissing in the picture in my brain?


Best,
Tobi

On Mon, Mar 2, 2020 at 1:36 PM Maximilian Michels <[email protected]<mailto:[email protected]>> wrote:


    Hi Tobi,

    That makes sense to me. My argument was coming from having
    "exactly-once" semantics for a pipeline. In this regard, the stop
    functionality does not help. But I think having the option to
    gracefully
    shut down a pipeline is beneficial for other uses cases like the ones
    you described.

    Could you create a JIRA issue? We can then target re-adding the stop
    functionality for the next release.

    Thanks,
    Max

    On 02.03.20 09:58, Kaymak, Tobias wrote:
     > Good morning Max and thanks for clarifying!
     >
     > I generated the JAR 2.19.0 in the second test via the default
    demo code
     > from Beam. There were no further adjustments from my side, but as
    I can
     > see there are some open points in JIRA for 1.9.2, so for now I think
     > that we can focus on 1.9.1 as a target.
     >
     > To understand the fail-safety paradigm correctly - It is my
     > understanding that, when I hit stop my pipeline (reading from Kafka,
     > writing to BigQuery) will stop consuming from Kafka and wait
    until all
     > bundles have been written out to BigQuery. Meaning that it will also
     > correctly persist the offset of the message it has successfully
    read in
     > Kafka. When it gets started again it will resume at that offset then.
     >
     > In contrast, hitting cancel causes my pipeline to drop everything
    it is
     > doing immediately and leaving me (without taking a snapshot) in an
     > unclear state.
     >
     > For operations it is useful to have a cancel button, to be able to
     > quickly shutdown a pipeline and completely re-read it from a
    streaming
     > source. For example when there was a critical bug in the pipeline
     > processing code, or when a team, that was sending data through that
     > source happened to release a feature earlier than announced and some
     > data had been missed, as the schema had not been updated in the
    pipeline
     > before.
     >
     > On the other hand, it is useful to have a stop button, to be able to
     > cleanly shutdown a pipeline when a feature is going to be released
     > tomorrow, and the pipeline processing code should be updated before,
     > without the overhead of re-reading the complete past. Taking a
    snapshot
     > uses the current representation of the Beam code as a Flink job, but
     > what if one wants to update the Beam code and thus the pipeline code
     > without the need to reprocess the whole history? Moreover, a stop
    button
     > is very useful when a new Flink version is going to be rolled
    out: then
     > one can drain all pipelines (for us there are right now 31),
    rollout a
     > new Flink version and start them at the point where they left of
    with
     > their last committed offset in Kafka.
     > Does that make sense?
     >
     > Best,
     > Tobi
     >
     > On Sun, Mar 1, 2020 at 5:23 PM Maximilian Michels <[email protected]
    <mailto:[email protected]>
     > <mailto:[email protected] <mailto:[email protected]>>> wrote:
     >
     >     In some sense, stop is different because sources will be
    stopped first
     >     and then all the downstream operators will stop
    automatically. However,
     >     in terms of correctness of your program using cancel or stop
    does not
     >     make a difference because neither approach takes a checkpoint.
     >
     >     Only at the time of a checkpoint you are guranteed to have a
    consistent
     >     state. A checkpoint is the only way to resume a canceled/stopped
     >     pipeline correctly.
     >
     >     Does that make sense? I understand that stop may have been
    convenient
     >     for your use case. If so, we may consider adding it again.
     >
     >     Cheers,
     >     Max
     >
     >     PS: Concerning the web interface with 1.9.2, I'm not sure
    what changes
     >     your Jar contain but we'll have to look into this when we
    upgrade to
     >     1.9.2 in Beam.
     >
     >     On 28.02.20 14:59, Kaymak, Tobias wrote:
     >      > I investigated further:
     >      >
     >      > As Flink 1.9.1 works and Flink 1.9.2 does not, I simply
    tried a
     >     rollout
     >      > of a vanilla flink:1.9.2-scala_2.11 image to K8s and that
    worked.
     >     So the
     >      > issue must be in my image or the JAR I am attaching:
     >      >
     >      > ARG FLINK_VERSION=1.9.2
     >      > ARG SCALA_VERSION=2.11
     >      > FROM flink:${FLINK_VERSION}-scala_${SCALA_VERSION}
     >      >
     >      > COPY --chown=flink:flink conf/log4j-console.properties
     >      > /opt/flink/conf/log4j-console.properties
     >      > ADD --chown=flink:flink
     >      >
     >
    
https://repo1.maven.org/maven2/org/apache/flink/flink-metrics-prometheus_${SCALA_VERSION}/${FLINK_VERSION}/flink-metrics-prometheus_${SCALA_VERSION}-${FLINK_VERSION}.jar
    
<https://repo1.maven.org/maven2/org/apache/flink/flink-metrics-prometheus_$%7BSCALA_VERSION%7D/$%7BFLINK_VERSION%7D/flink-metrics-prometheus_$%7BSCALA_VERSION%7D-$%7BFLINK_VERSION%7D.jar>

> <https://repo1.maven.org/maven2/org/apache/flink/flink-metrics-prometheus_$%7BSCALA_VERSION%7D/$%7BFLINK_VERSION%7D/flink-metrics-prometheus_$%7BSCALA_VERSION%7D-$%7BFLINK_VERSION%7D.jar>

     >
     >      >

> /opt/flink/lib/flink-metrics-prometheus_${SCALA_VERSION}-${FLINK_VERSION}.jar

     >      > ADD --chown=flink:flink
     >      >
     >
    
https://repo1.maven.org/maven2/org/apache/flink/flink-statebackend-rocksdb_${SCALA_VERSION}/${FLINK_VERSION}/flink-statebackend-rocksdb_${SCALA_VERSION}-${FLINK_VERSION}.jar
    
<https://repo1.maven.org/maven2/org/apache/flink/flink-statebackend-rocksdb_$%7BSCALA_VERSION%7D/$%7BFLINK_VERSION%7D/flink-statebackend-rocksdb_$%7BSCALA_VERSION%7D-$%7BFLINK_VERSION%7D.jar>

> <https://repo1.maven.org/maven2/org/apache/flink/flink-statebackend-rocksdb_$%7BSCALA_VERSION%7D/$%7BFLINK_VERSION%7D/flink-statebackend-rocksdb_$%7BSCALA_VERSION%7D-$%7BFLINK_VERSION%7D.jar>

     >
     >      >

> /opt/flink/lib/flink-statebackend-rocksdb_${SCALA_VERSION}-${FLINK_VERSION}.jar

     >      > ADD --chown=flink:flink deployment/run.sh /opt/flink/run.sh
     >      > RUN chmod +x /opt/flink/run.sh
     >      >
     >      > COPY --from=builder --chown=flink:flink
     >      > /build/target/di-beam-bundled.jar
    /opt/flink/lib/beam_pipelines.jar
     >      >
     >      > Commenting out the last COPY step - to circumvent the
    addition of
     >     the
     >      > fat Beam JAR - did the trick. So I think my beam JAR contains
     >     something
     >      > Flink does not like. Next thing I tried was building a vanilla
     >     Beam JAR via:
     >      >
     >      > mvn archetype:generate \
     >      >        -DarchetypeGroupId=org.apache.beam \
     >      >
     >     -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
     >      >        -DarchetypeVersion=2.19.0 \
     >      >        -DgroupId=org.example \
     >      >        -DartifactId=word-count-beam \
     >      >        -Dversion="0.1" \
     >      >        -Dpackage=org.apache.beam.examples \
     >      >        -DinteractiveMode=false
     >      >
     >      > and then mvn package -Pflink-runner - even with this JAR the
     >      > webinterface of the jobmanager does not come up.
     >      >
     >      > Best,
     >      > Tobias
     >      >
     >      > On Fri, Feb 28, 2020 at 11:36 AM Kaymak, Tobias
     >      > <[email protected]
    <mailto:[email protected]> <mailto:[email protected]
    <mailto:[email protected]>>
     >     <mailto:[email protected]
    <mailto:[email protected]> <mailto:[email protected]
    <mailto:[email protected]>>>>
     >     wrote:
     >      >
     >      >     Hi Max,
     >      >
     >      >     I am surprised that Stop and Cancel had the same action in
     >     Flink 1.8
     >      >     and lower. When I clicked on stop in the past the
    whole pipeline
     >      >     shutdown each operator one after the other and after a
    couple of
     >      >     minutes it was done and all outstanding bundles were
    imported
     >      >     correctly (my pipeline goes from Kafka to BigQuery).
    When I
     >     clicked
     >      >     cancel it stopped immediately, leaving some data
    (bundles) in the
     >      >     "air". Was my understanding wrong? From an operational
    point
     >     of view
     >      >     when clicking cancel and starting the pipeline again
    will it
     >     resume
     >      >     without a loss of data?
     >      >
     >      >     Here is my mail to the Flink mailing list explaining
    my issue
     >     with
     >      >     1.9.2:
     >      >
     >      >     I enabled DEBUG/TRACE logging and it looks ok to me when I
     >     make the
     >      >     request after the jobmanager pod has been started:
     >      >
     >      >     2020-02-27 10:11:42,439 TRACE