Hi everyone, I've done some digging within the Beam source code. It looks like when the flinkMaster argument is not set, the savepointPath is not used at all. [1]
In fact the only time the savepointPath argument is used within all of Beam's source code is on lines 183 and 186 of the same file. [2] Of course, I did all my testing locally on my dev box with the embedded Flink cluster that Beam starts, which from the looks of it, does NOT use the savepointPath at all. If someone familiar with the code can confirm this finding, I can update the documentation to explicitly state that savepoint resuming is not supported locally. I will do more testing around this with a real Flink cluster and see if the behavior is different than the one described in my first email. Thanks, Cristian [1] https://github.com/apache/beam/blob/51e0e4ef82feaae251b37a0288ad7a04ee603086/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java#L174 [2] https://github.com/apache/beam/blob/51e0e4ef82feaae251b37a0288ad7a04ee603086/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java#L183 On Wed, Feb 2, 2022 at 3:01 PM Cristian Constantinescu <[email protected]> wrote: > Hey Pavel, > > Thanks for the quick reply. Pardon me as I cannot copy/paste straight from > the IDE, copying by hand: > > KafkaIO.<Pojo>read() > .withBootStrapServer("address") > .withTopic("topic") > > .withKeyDeserializer(StringDeserializer.class) > > .withValueDeserializer(ConfluentSchemaRegistryDesProvider.of(...)) > > .withConsumerConfigUpdates(map) > .withReadCommitted() > .commitOffsetInFinalize() > > .withProcessingTime(); > > > The config map is: > enable.auto.commit -> false > group.id -> some group > auto.offset.reset -> earliest > specific.avro.reader -> false > > > On Wed, Feb 2, 2022 at 2:44 PM Pavel Solomin <[email protected]> > wrote: > >> Hello Christian, >> >> Thanks for posting here the detailed scenario of your experiments. I >> think it may be important to share your KafkaIO configuration here too. For >> example, are you setting this config anyhow? >> https://beam.apache.org/releases/javadoc/2.35.0/org/apache/beam/sdk/io/kafka/KafkaIO.Read.html#commitOffsetsInFinalize-- >> >> Best Regards, >> Pavel Solomin >> >> Tel: +351 962 950 692 | Skype: pavel_solomin | Linkedin >> <https://www.linkedin.com/in/pavelsolomin> >> >> >> >> >> >> On Wed, 2 Feb 2022 at 19:20, Cristian Constantinescu <[email protected]> >> wrote: >> >>> Hi everyone, >>> >>> I'm trying to figure out how pipeline state works with Beam running on >>> Flink Classic. Would appreciate some help with the below. >>> >>> My understanding is that on recovery (whether from a checkpoint or >>> savepoint), Flink recreates the operators (I guess those are DoFns in Beam) >>> with whatever state they had when the pipeline crashed. For example the >>> Kafka operator might contain the latest *safe* offset to restart from. But >>> I'm not seeing this when I introduce exceptions in the pipeline. >>> >>> My pipeline is as follows: >>> 1. Read a Kafka topic from start >>> 2. Have a DoFn that stores all incoming messages in a BagState >>> 3. Above DoFn triggers a timer set in such a way that it triggers after >>> there are a few checkpoints created and kept because of >>> --externalizeCheckpointsEnabled = true. This timer handler then outputs the >>> elements to the next operator, in this case KafkaIo.Write. >>> 4. Before the timer in #3 is executed manually trigger an exception >>> (listen to another kafka topic, and throw any time a new message comes in) >>> >>> What I observe: >>> 1. In #4 above Flink tries to process the exception twice then stops the >>> pipeline (because numberOfExecutionRetries =2 ) >>> 2. After the pipeline is stopped, I see the checkpoints are kept in the >>> configured directory >>> 3. If I restart the pipeline (with --savepointPath = <path to latest >>> checkpoint from first run>): >>> 3a. No messages are read from kafka, because the Kafka reader reached >>> the end of the topic during the first run >>> 3b. StartBundles are not executed for my DoFn. Indicating that the DoFn >>> isn't even started >>> 3c. The timer in #3 is never executed, hence there is data loss as the >>> elements I had in my DoFn state are never processed >>> 4. If I manually reset the offset to the start of the topic and restart >>> the pipeline (with --savepointPath = <path to latest checkpoint from first >>> run>): >>> 4a. StartBundle methods are called >>> 4b. In ProcessElement, the BagState is empty on the first received >>> message. If I'm restoring from a checkpoint/savepoint, I would expect this >>> state to be filled. >>> >>> Is this correct behaviour? Am I doing something wrong? >>> >>> Thanks, >>> Cristian >>> >>> Other quirks I found: >>> a. If KafkaIO.Read is configured to read from the latest offset, and >>> there is an exception thrown in the pipeline before the first checkpoint >>> happens (let's say on the first message that comes in), when Flink retries >>> KafkaIO reads from the latest offset again. That means that the message >>> that caused the exception is not reprocessed. On the other hand, if the >>> exception is thrown after the first checkpoint, that message will be tried >>> twice (because numberOfExecutionRetries =2 ), and then the pipeline will >>> exit. I think this is working as designed but it feels a little weird >>> that the behaviour is different depending if there's a checkpoint or not. >>> >>> b. When KafkaIO.Write is configured with .withEOS(number, "group"), and >>> there is an exception thrown in the pipeline, the Flink job doesn't exit. I >>> think there is a kafka producer in KafkaExactlyOnceSink that is not closed >>> correctly. >>> >>
