> do you mean inside the processElement() method? I used a simple mapper with Thread.sleep before ExceptionSimulatorProcess.
> what is 0ms pause? do you mean > env.getCheckpointConfig().setMinPauseBetweenCheckpoints(0); ? Yes, env.getCheckpointConfig().setMinPauseBetweenCheckpoints(0); > How do you create a random exception? do you mean not mine SimulatedException? I mean it should be thrown at random because the checkpoint must reliably precede it. So on recovery that there is some state. Checking against 6666... only once assumes that the checkpoint was triggered before. Besides, checkpoint is not guaranteed to be triggered before the end of input. I tried to run it in with maven and it worked after making the source infinite. >From the code you provided the parallelism level doesn't seem important and can be set 1 (or restart strategy to full). Then using getRuntimeContext().getAttemptNumber() would be simpler and more reliable. Regards, Roman On Fri, Jun 18, 2021 at 6:23 PM Felipe Gutierrez <felipe.o.gutier...@gmail.com> wrote: > > > > On Fri, Jun 18, 2021 at 5:40 PM Roman Khachatryan <ro...@apache.org> wrote: >> >> I tried to run the test that you mentioned >> (WordCountFilterQEPTest#integrationTestWithPoisonPillRecovery) in rev. >> 6f08d0a. >> >> In IDE, I see that: >> - checkpoint is never triggered (sentence is too short, checkpoint >> pause and interval are too large) >> - exception is never thrown, so the job never restarted >> (currentTimeMillis is incremented but referenceTimeMillisAhead is not) >> >> When I add sleep between each element, set 10ms interval, 0ms pause >> and introduce some random exception, I do see > > > do you mean inside the processElement() method? > > what is 0ms pause? do you mean > env.getCheckpointConfig().setMinPauseBetweenCheckpoints(0); ? > > How do you create a random exception? do you mean not mine SimulatedException? > > Using these configurations that I just said it is not working for me. I am > testing on the terminal "mvn > -Dtest=WordCountFilterQEPTest#integrationTestWithPoisonPillRecovery test". On > IntelliJ I have the error "Caused by: java.lang.NoSuchFieldError: > latencyTrackingConfigBuilder" when I call "env.execute();" > > org.apache.flink.runtime.client.JobExecutionException: Job execution failed. > at > org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147) > at > org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$2(MiniClusterJobClient.java:119) > at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616) > at > java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:628) > at > java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:1996) > at > org.apache.flink.runtime.minicluster.MiniClusterJobClient.getJobExecutionResult(MiniClusterJobClient.java:117) > at > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1842) > at > org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:70) > at > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1822) > at > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1804) > at > org.sense.flink.examples.stream.edgent.WordCountFilterQEPTest.integrationTestWithPoisonPillRecovery(WordCountFilterQEPTest.java:210) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > at > org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59) > at > org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) > at > org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56) > at > org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) > at > org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103) > at > org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63) > at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) > at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at org.junit.runners.ParentRunner.run(ParentRunner.java:413) > at org.junit.runner.JUnitCore.run(JUnitCore.java:137) > at > com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69) > at > com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33) > at > com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:221) > at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54) > Caused by: java.lang.NoSuchFieldError: latencyTrackingConfigBuilder > > > >> >> 2021-06-18 17:37:53,241 INFO >> org.sense.flink.examples.stream.edgent.ExceptionSimulatorProcess - >> Attempts restart: 1 >> in the logs. >> >> These settings probably differ on the cluster and there is some >> unrelated exception which causes a restart. >> >> Regards, >> Roman >> >> On Fri, Jun 18, 2021 at 12:20 PM Felipe Gutierrez >> <felipe.o.gutier...@gmail.com> wrote: >> > >> > I investigated a little bit more. I created the same POC on a Flink >> > version 1.13. I have this ProcessFunction where I want to count the times >> > it recovers. I tested with ListState and ValueState and it seems that >> > during the integration test (only for integration test) the process is >> > hanging on the open() or on the initializeState() methods. >> > >> > https://github.com/felipegutierrez/explore-flink/blob/master/docker/ops-playground-image/java/explore-flink/src/main/java/org/sense/flink/examples/stream/edgent/ExceptionSimulatorProcess.java >> > >> > Here is my integration test: >> > https://github.com/felipegutierrez/explore-flink/blob/master/docker/ops-playground-image/java/explore-flink/src/test/java/org/sense/flink/examples/stream/edgent/WordCountFilterQEPTest.java#L154 >> > >> > If I comment out the state ListState or the ValueState, the integration >> > test works. It first throws an exception and I see it on the output. Then >> > after 2 seconds it does not throw exceptions. Then I compare the output >> > and they match. But I don't have a way to count how many restarts happened >> > during the integration test. I con only count the restarts when I run the >> > application on a stand-alone flink cluster. >> > >> > I don't know. Maybe, do I have to configure some parameters to work with >> > the state on integration tests? >> > >> > -- >> > -- Felipe Gutierrez >> > -- skype: felipe.o.gutierrez >> > >> > >> > On Fri, Jun 18, 2021 at 9:31 AM Felipe Gutierrez >> > <felipe.o.gutier...@gmail.com> wrote: >> >> >> >> No, it didn't work. >> >> >> >> The "context.isRestored()" returns true when I run the application on the >> >> Flink standalone-cluster and it is recovering after a failure. When I do >> >> the same on a integration test it does not returns true after a failure. >> >> I mean, I can log the exception that is causing the failure, the >> >> initializeState() is called after a failure, but the context.isRestored() >> >> is false again. I also tried to update the state on the first time to 0 >> >> "if (!context.isRestored()) { restartsState.add(0L); }" and it does not >> >> work. >> >> I think the problem is not on the ListState that I am using and not on >> >> the context.isRestore() as well. It is on the >> >> "context.getOperatorStateStore()" that is always null only on integration >> >> tests. Using the below code I can see on the logs "restarts: 0" twice, >> >> before and after failure. >> >> >> >> @Override >> >> public void initializeState(FunctionInitializationContext context) >> >> throws Exception { >> >> // unit tests does not open OperatorStateStore >> >> if (context.getOperatorStateStore() != null) { >> >> restartsState = >> >> context.getOperatorStateStore().getListState(new >> >> ListStateDescriptor<Long>("restarts", Long.class)); >> >> >> >> List<Long> restoreList = >> >> Lists.newArrayList(restartsState.get()); >> >> if (restoreList == null || restoreList.isEmpty()) { >> >> restartsState.add(0L); >> >> LOG.info("restarts: 0"); >> >> } else { >> >> Long max = Collections.max(restoreList); >> >> LOG.info("restarts: " + max); >> >> restartsState.add(max + 1); >> >> } >> >> } >> >> } >> >> >> >> -- >> >> -- Felipe Gutierrez >> >> -- skype: felipe.o.gutierrez >> >> >> >> >> >> On Thu, Jun 17, 2021 at 11:17 PM Roman Khachatryan <ro...@apache.org> >> >> wrote: >> >>> >> >>> Thanks for sharing, >> >>> >> >>> I think the problem is that restartsState is never updated: >> >>> - on the first attempt, context.isRestored() returns false (and "never >> >>> restored" is logged) >> >>> - on subsequent attempts, it again returns false, because the state >> >>> was never updated before >> >>> >> >>> Adding >> >>> if (!context.isRestored()) { restartsState.add(0L); } >> >>> should solve the problem >> >>> (it's also better to use state.update instead of state.add if only max >> >>> is needed). >> >>> >> >>> Regards, >> >>> Roman >> >>> >> >>> On Thu, Jun 17, 2021 at 5:35 PM Felipe Gutierrez >> >>> <felipe.o.gutier...@gmail.com> wrote: >> >>> > >> >>> > Sure, here it is. Nothing is mocked. I double-checked. >> >>> > >> >>> > UnitTestClass {..... >> >>> > protected static LocalFlinkMiniCluster flink; >> >>> > >> >>> > @BeforeClass >> >>> > public static void prepare() { >> >>> > flink = new LocalFlinkMiniCluster(getFlinkConfiguration(), false); >> >>> > flink.start(); >> >>> > >> >>> > TestStreamEnvironment.setAsContext(flink, PARALLELISM); >> >>> > } >> >>> > >> >>> > private static Configuration getFlinkConfiguration() { >> >>> > Configuration flinkConfig = new Configuration(); >> >>> > flinkConfig.setInteger("local.number-taskmanager", 1); >> >>> > flinkConfig.setInteger("taskmanager.numberOfTaskSlots", 8); >> >>> > flinkConfig.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 16L); >> >>> > flinkConfig.setString("restart-strategy.fixed-delay.delay", "0 s"); >> >>> > try { >> >>> > flinkConfig.setString("state.checkpoints.dir", "file://" + >> >>> > tempFolder.newFolder().getAbsolutePath()); >> >>> > } catch (IOException e) { >> >>> > throw new RuntimeException("error in flink cluster config", e); >> >>> > } >> >>> > return flinkConfig; >> >>> > } >> >>> > >> >>> > >> >>> > The class that I check if the job was restarted: >> >>> > >> >>> > public class ExceptionSimulatorProcessFunction extends >> >>> > ProcessFunction<Object..., Object...> >> >>> > implements CheckpointedFunction { >> >>> > >> >>> > final OutputTag<Long> outputTag = new >> >>> > OutputTag<Long>("side-output") { >> >>> > }; >> >>> > private transient ListState<Long> restartsState; >> >>> > private Long restartsLocal; >> >>> > ... >> >>> > @Override >> >>> > public void processElement(Object value, Context ctx, >> >>> > Collector<Object> out) throws Exception { >> >>> > this.currentTimeMillis = System.currentTimeMillis() - >> >>> > currentTimeMillisBehind; >> >>> > >> >>> > // If current time is less than the reference time ahead AND >> >>> > we have the poison auction an exception will throw >> >>> > if (this.currentTimeMillis < this.referenceTimeMillisAhead && >> >>> > POISON__TRANSACTION_ID.equals(value.toString())) { >> >>> > >> >>> > LOG.error("This exception will trigger until the reference >> >>> > time [{}] reaches the trigger time [{}]", >> >>> > sdfMillis.format(new Date(this.currentTimeMillis)), >> >>> > sdfMillis.format(new >> >>> > Date(this.referenceTimeMillisAhead))); >> >>> > >> >>> > throw new SimulatedException("Transaction ID: " + >> >>> > value.toString() + >> >>> > " not allowed. This is a simple exception for >> >>> > testing purposes."); >> >>> > } >> >>> > out.collect(value); >> >>> > >> >>> > >> >>> > // counts the restarts >> >>> > if (restartsState != null) { >> >>> > List<Long> restoreList = >> >>> > Lists.newArrayList(restartsState.get()); >> >>> > Long attemptsRestart = 0L; >> >>> > if (restoreList != null && !restoreList.isEmpty()) { >> >>> > attemptsRestart = Collections.max(restoreList); >> >>> > if (restartsLocal < attemptsRestart) { >> >>> > restartsLocal = attemptsRestart; >> >>> > ctx.output(outputTag, >> >>> > Long.valueOf(attemptsRestart)); >> >>> > } >> >>> > } >> >>> > LOG.info("Attempts restart: " + attemptsRestart); >> >>> > } >> >>> > } >> >>> > >> >>> > @Override >> >>> > public void snapshotState(FunctionSnapshotContext context) throws >> >>> > Exception {} >> >>> > >> >>> > @Override >> >>> > public void initializeState(FunctionInitializationContext context) >> >>> > throws Exception { >> >>> > restartsState = >> >>> > context.getOperatorStateStore().getListState(new >> >>> > ListStateDescriptor<Long>("restarts", Long.class)); >> >>> > >> >>> > if (context.isRestored()) { >> >>> > List<Long> restoreList = >> >>> > Lists.newArrayList(restartsState.get()); >> >>> > if (restoreList == null || restoreList.isEmpty()) { >> >>> > restartsState.add(1L); >> >>> > LOG.info("restarts: 1"); >> >>> > } else { >> >>> > Long max = Collections.max(restoreList); >> >>> > LOG.info("restarts: " + max); >> >>> > restartsState.add(max + 1); >> >>> > } >> >>> > } else { >> >>> > LOG.info("restarts: never restored"); >> >>> > } >> >>> > } >> >>> > } >> >>> > >> >>> > >> >>> > >> >>> > >> >>> > >> >>> > >> >>> > >> >>> > >> >>> > On Thu, Jun 17, 2021 at 5:20 PM Roman Khachatryan <ro...@apache.org> >> >>> > wrote: >> >>> >> >> >>> >> Hi, >> >>> >> >> >>> >> Could you please share the test code? >> >>> >> >> >>> >> I think the returned value might depend on the level on which the >> >>> >> tests are executed. If it's a regular job then it should return the >> >>> >> correct value (as with cluster). If the environment in which the code >> >>> >> is executed is mocked then it can be false. >> >>> >> >> >>> >> Regards, >> >>> >> Roman >> >>> >> >> >>> >> On Thu, Jun 17, 2021 at 4:15 PM Felipe Gutierrez >> >>> >> <felipe.o.gutier...@gmail.com> wrote: >> >>> >> > >> >>> >> > Yes, I have state on the ProcessFunction. I tested it on a >> >>> >> > stand-alone cluster and it returns true when the application >> >>> >> > recovers. However, in integration tests it does not returns true. I >> >>> >> > am using Flink 1.4. Do you know where it is saying at Flink release >> >>> >> > 1.13 (https://flink.apache.org/news/2021/05/03/release-1.13.0.html) >> >>> >> > that I cannot see `isRestored()` equals true on integration tests? >> >>> >> > >> >>> >> > -- >> >>> >> > -- Felipe Gutierrez >> >>> >> > -- skype: felipe.o.gutierrez >> >>> >> > >> >>> >> > >> >>> >> > On Thu, Jun 17, 2021 at 4:09 PM Arvid Heise <ar...@apache.org> >> >>> >> > wrote: >> >>> >> >> >> >>> >> >> Does your ProcessFunction has state? If not it would be in line >> >>> >> >> with the documentation. >> >>> >> >> >> >>> >> >> Also which Flink version are you using? Before Flink 1.13 empty >> >>> >> >> state was omitted so I could imagine that `isRestored()` would >> >>> >> >> return false but it should actually now also return true for empty >> >>> >> >> state. >> >>> >> >> >> >>> >> >> On Tue, Jun 15, 2021 at 9:02 AM Felipe Gutierrez >> >>> >> >> <felipe.o.gutier...@gmail.com> wrote: >> >>> >> >>> >> >>> >> >>> So, I was trying to improve by using the CheckpointedFunction as >> >>> >> >>> it shows here [1]. But the method isRestored() says in its >> >>> >> >>> documentation [2]: >> >>> >> >>> >> >>> >> >>> "Returns true, if state was restored from the snapshot of a >> >>> >> >>> previous execution. This returns always false for stateless >> >>> >> >>> tasks." >> >>> >> >>> >> >>> >> >>> It is weird because I am extending a ProcessFunction which is a >> >>> >> >>> RichFunction. >> >>> >> >>> >> >>> >> >>> public class AuctionExceptionSimulatorProcessFunction extends >> >>> >> >>> ProcessFunction<KeyedReportingData, KeyedReportingData> >> >>> >> >>> implements CheckpointedFunction { >> >>> >> >>> ... >> >>> >> >>> >> >>> >> >>> In the end, I cannot rely on the "isRestored()". Do you know what >> >>> >> >>> could be wrong? I used the same implementation method of [1]. >> >>> >> >>> >> >>> >> >>> [1] >> >>> >> >>> https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/datastream/fault-tolerance/state/#checkpointedfunction >> >>> >> >>> [2] >> >>> >> >>> https://ci.apache.org/projects/flink/flink-docs-release-1.4/api/java/org/apache/flink/runtime/state/ManagedInitializationContext.html#isRestored-- >> >>> >> >>> >> >>> >> >>> >> >>> >> >>> -- >> >>> >> >>> -- Felipe Gutierrez >> >>> >> >>> -- skype: felipe.o.gutierrez >> >>> >> >>> >> >>> >> >>> >> >>> >> >>> On Mon, Jun 14, 2021 at 3:39 PM Roman Khachatryan >> >>> >> >>> <ro...@apache.org> wrote: >> >>> >> >>>> >> >>> >> >>>> You can also use accumulators [1] to collect the number of >> >>> >> >>>> restarts >> >>> >> >>>> (and then access it via client); but side outputs should work as >> >>> >> >>>> well. >> >>> >> >>>> >> >>> >> >>>> [1] >> >>> >> >>>> https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/api_concepts.html#accumulators--counters >> >>> >> >>>> >> >>> >> >>>> Regards, >> >>> >> >>>> Roman >> >>> >> >>>> >> >>> >> >>>> On Sun, Jun 13, 2021 at 10:01 PM Felipe Gutierrez >> >>> >> >>>> <felipe.o.gutier...@gmail.com> wrote: >> >>> >> >>>> > >> >>> >> >>>> > I just realised that only the ProcessFunction is enough. I >> >>> >> >>>> > don't need the CheckpointFunction. >> >>> >> >>>> > >> >>> >> >>>> > >> >>> >> >>>> > On Fri, 11 Jun 2021, 18:11 Felipe Gutierrez, >> >>> >> >>>> > <felipe.o.gutier...@gmail.com> wrote: >> >>> >> >>>> >> >> >>> >> >>>> >> Cool! >> >>> >> >>>> >> >> >>> >> >>>> >> I did using this example >> >>> >> >>>> >> https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/stream/state/state.html#using-managed-operator-state >> >>> >> >>>> >> because I don't have a keyed stream on the specific operator >> >>> >> >>>> >> that I want to count the number of restarts. (yes I am using >> >>> >> >>>> >> version 1.4 unfortunately). >> >>> >> >>>> >> >> >>> >> >>>> >> Because I need to test it in an integration test I am using a >> >>> >> >>>> >> side output >> >>> >> >>>> >> (https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/stream/testing.html) >> >>> >> >>>> >> to attach a sink. I am not sure if you have a better idea to >> >>> >> >>>> >> test the restarts on an integration test. If you have a >> >>> >> >>>> >> simple idea please tell me :). This was the way that I >> >>> >> >>>> >> solved.... >> >>> >> >>>> >> >> >>> >> >>>> >> Thanks >> >>> >> >>>> >> Felipe >> >>> >> >>>> >> >> >>> >> >>>> >> -- >> >>> >> >>>> >> -- Felipe Gutierrez >> >>> >> >>>> >> -- skype: felipe.o.gutierrez >> >>> >> >>>> >> >> >>> >> >>>> >> >> >>> >> >>>> >> On Thu, Jun 10, 2021 at 5:41 PM Roman Khachatryan >> >>> >> >>>> >> <ro...@apache.org> wrote: >> >>> >> >>>> >>> >> >>> >> >>>> >>> Hi Felipe, >> >>> >> >>>> >>> >> >>> >> >>>> >>> You can use getRuntimeContext().getAttemptNumber() [1] (but >> >>> >> >>>> >>> beware >> >>> >> >>>> >>> that depending on the configuration only a pipeline region >> >>> >> >>>> >>> can be >> >>> >> >>>> >>> restarted, not the whole job). >> >>> >> >>>> >>> >> >>> >> >>>> >>> But if all you want is to check whether it's a first attempt >> >>> >> >>>> >>> or not, >> >>> >> >>>> >>> you can also call context.isRestored() from >> >>> >> >>>> >>> initializeState() [2] >> >>> >> >>>> >>> >> >>> >> >>>> >>> [1] >> >>> >> >>>> >>> https://ci.apache.org/projects/flink/flink-docs-release-1.13/api/java/org/apache/flink/api/common/functions/RuntimeContext.html#getAttemptNumber-- >> >>> >> >>>> >>> >> >>> >> >>>> >>> [2] >> >>> >> >>>> >>> https://ci.apache.org/projects/flink/flink-docs-release-1.13/api/java/org/apache/flink/runtime/state/ManagedInitializationContext.html#isRestored-- >> >>> >> >>>> >>> >> >>> >> >>>> >>> Regards, >> >>> >> >>>> >>> Roman >> >>> >> >>>> >>> >> >>> >> >>>> >>> >> >>> >> >>>> >>> On Thu, Jun 10, 2021 at 5:30 PM Felipe Gutierrez >> >>> >> >>>> >>> <felipe.o.gutier...@gmail.com> wrote: >> >>> >> >>>> >>> > >> >>> >> >>>> >>> > Hello community, >> >>> >> >>>> >>> > >> >>> >> >>>> >>> > Is it possible to know programmatically how many times my >> >>> >> >>>> >>> > Flink stream job restarted since it was running? >> >>> >> >>>> >>> > >> >>> >> >>>> >>> > My use case is like this. I have an Unit test that uses >> >>> >> >>>> >>> > checkpoint and I throw one exception in a MapFunction for >> >>> >> >>>> >>> > a given time, i.e.: for the 2 seconds ahead. Because Flink >> >>> >> >>>> >>> > restarts the job and I have checkpoint I can recover the >> >>> >> >>>> >>> > state and after 2 seconds I don't throw any exception >> >>> >> >>>> >>> > anymore. Then I would like to know how many times the job >> >>> >> >>>> >>> > was restarted. >> >>> >> >>>> >>> > >> >>> >> >>>> >>> > Thanks, >> >>> >> >>>> >>> > Felipe >> >>> >> >>>> >>> >
