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 04:36:28 -0800

Hi Tobi,

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

Could you create a JIRA issue? We can then target re-adding the stopfunctionality 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 codefrom Beam. There were no further adjustments from my side, but as I cansee there are some open points in JIRA for 1.9.2, so for now I thinkthat we can focus on 1.9.1 as a target.

To understand the fail-safety paradigm correctly - It is myunderstanding that, when I hit stop my pipeline (reading from Kafka,writing to BigQuery) will stop consuming from Kafka and wait until allbundles have been written out to BigQuery. Meaning that it will alsocorrectly persist the offset of the message it has successfully read inKafka. When it gets started again it will resume at that offset then.

In contrast, hitting cancel causes my pipeline to drop everything it isdoing immediately and leaving me (without taking a snapshot) in anunclear state.

For operations it is useful to have a cancel button, to be able toquickly shutdown a pipeline and completely re-read it from a streamingsource. For example when there was a critical bug in the pipelineprocessing code, or when a team, that was sending data through thatsource happened to release a feature earlier than announced and somedata had been missed, as the schema had not been updated in the pipelinebefore.

On the other hand, it is useful to have a stop button, to be able tocleanly shutdown a pipeline when a feature is going to be releasedtomorrow, and the pipeline processing code should be updated before,without the overhead of re-reading the complete past. Taking a snapshotuses the current representation of the Beam code as a Flink job, butwhat if one wants to update the Beam code and thus the pipeline codewithout the need to reprocess the whole history? Moreover, a stop buttonis very useful when a new Flink version is going to be rolled out: thenone can drain all pipelines (for us there are right now 31), rollout anew Flink version and start them at the point where they left of withtheir 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]>> 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>

     >
    
/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>

     >
    
/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]>>>
    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
     >     org.apache.flink.runtime.rest.FileUploadHandler               -
     >     Received request. URL:/ Method:GET
     >     2020-02-27 10:11:42,440 DEBUG