date:20231215

[jira] [Updated] (FLINK-33862) Flink Unit Test Failures on 1.18.0

2023-12-15 Thread Prabhu Joseph (Jira)



 [ 
https://issues.apache.org/jira/browse/FLINK-33862?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Prabhu Joseph updated FLINK-33862:
--
Affects Version/s: 1.19.0

> Flink Unit Test Failures on 1.18.0
> --
>
> Key: FLINK-33862
> URL: https://issues.apache.org/jira/browse/FLINK-33862
> Project: Flink
>  Issue Type: Bug
>Affects Versions: 1.18.0, 1.19.0
>Reporter: Prabhu Joseph
>Priority: Major
>
> Flink Unit Test Failures on 1.18.0. There are 100+ unit test cases failing 
> due to below common issues.
> *Issue 1*
> {code:java}
> ./mvnw -DfailIfNoTests=false -Dmaven.test.failure.ignore=true 
> -Dtest=ExecutionPlanAfterExecutionTest test
> [INFO] Running org.apache.flink.client.program.ExecutionPlanAfterExecutionTest
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
>   at 
> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
>   at 
> org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:141)
>   at 
> java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:642)
>   at 
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
>   at 
> java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
>   at 
> org.apache.flink.runtime.rpc.pekko.PekkoInvocationHandler.lambda$invokeRpc$1(PekkoInvocationHandler.java:268)
>   at 
> java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
>   at 
> java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
>   at 
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
>   at 
> java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
>   at 
> org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1267)
>   at 
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$1(ClassLoadingUtils.java:93)
>   at 
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
>   at 
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)
>   at 
> java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
>   at 
> java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
>   at 
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
>   at 
> java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
>   at 
> org.apache.flink.runtime.concurrent.pekko.ScalaFutureUtils$1.onComplete(ScalaFutureUtils.java:47)
>   at org.apache.pekko.dispatch.OnComplete.internal(Future.scala:310)
>   at org.apache.pekko.dispatch.OnComplete.internal(Future.scala:307)
>   at org.apache.pekko.dispatch.japi$CallbackBridge.apply(Future.scala:234)
>   at org.apache.pekko.dispatch.japi$CallbackBridge.apply(Future.scala:231)
>   at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
>   at 
> org.apache.flink.runtime.concurrent.pekko.ScalaFutureUtils$DirectExecutionContext.execute(ScalaFutureUtils.java:65)
>   at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:72)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:288)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:288)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:288)
>   at org.apache.pekko.pattern.PromiseActorRef.$bang(AskSupport.scala:629)
>   at 
> org.apache.pekko.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:34)
>   at 
> org.apache.pekko.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:33)
>   at scala.concurrent.Future.$anonfun$andThen$1(Future.scala:536)
>   at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
>   at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
>   at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
>   at 
> org.apache.pekko.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:73)
>   at 
> org.apache.pekko.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:110)
>   at 
> scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
>   at 
> scala.concurrent.BlockContext$.with

Re: [PR] [FLINK-33863] Fix restoring compressed operator state [flink]

2023-12-15 Thread via GitHub



flinkbot commented on PR #23938:
URL: https://github.com/apache/flink/pull/23938#issuecomment-1858739524

   
   ## CI report:
   
   * 052dab59db8f519765c51228f87277bd85b83a45 UNKNOWN
   
   
   Bot commands
 The @flinkbot bot supports the following commands:
   
- `@flinkbot run azure` re-run the last Azure build
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[jira] [Updated] (FLINK-33863) Compressed Operator state restore failed

2023-12-15 Thread ASF GitHub Bot (Jira)



 [ 
https://issues.apache.org/jira/browse/FLINK-33863?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

ASF GitHub Bot updated FLINK-33863:
---
Labels: pull-request-available  (was: )

> Compressed Operator state restore failed
> 
>
> Key: FLINK-33863
> URL: https://issues.apache.org/jira/browse/FLINK-33863
> Project: Flink
>  Issue Type: Bug
>  Components: Runtime / State Backends
>Affects Versions: 1.18.0
>Reporter: Ruibin Xing
>Priority: Major
>  Labels: pull-request-available
>
> We encountered an issue when using Flink 1.18.0. Our job enabled Snapshot 
> Compression and used multiple Operator States in an operator. When recovering 
> Operator State from a Savepoint, the following error occurred: 
> "org.xerial.snappy.SnappyFramedInputStream: encountered EOF while reading 
> stream header."
> After researching, I believe the error is due to Flink 1.18.0's support for 
> Snapshot Compression on Operator State (see 
> https://issues.apache.org/jira/browse/FLINK-30113 ). When writing a 
> Savepoint, SnappyFramedInputStream adds a header to the beginning of the 
> data. When recovering Operator State from a Savepoint, 
> SnappyFramedInputStream verifies the header from the beginning of the data.
> Currently, when recovering Operator State with Snapshot Compression enabled, 
> the logic is as follows:
> For each OperatorStateHandle:
> 1. Verify if the current Savepoint stream's offset is the Snappy header.
> 2. Seek to the state's start offset.
> 3. Read the state's data and finally seek to the state's end offset.
> (See: 
> https://github.com/apache/flink/blob/ef2b626d67147797e992ec3b338bafdb4e5ab1c7/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java#L172
>  )
> Furthermore, when there are multiple Operator States, they are not sorted 
> according to the Operator State's offset. Therefore, if the Operator States 
> are out of order and the final offset is recovered first, the Savepoint 
> stream will be seeked to the end, resulting in an EOF error.
> I propose a solution: sort the OperatorStateHandle by offset and then recover 
> the Operator State in order. After testing, this approach resolves the issue.
> I will submit a PR. This is my first time contributing code, so any help is 
> really appreciated.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[PR] [FLINK-33863] Fix restoring compressed operator state [flink]

2023-12-15 Thread via GitHub



ruibinx opened a new pull request, #23938:
URL: https://github.com/apache/flink/pull/23938

   ## What is the purpose of the change
   
   This PR fixes an issue related to restoring multiple operator states when 
snapshot compression is enabled. see: 
https://issues.apache.org/jira/browse/FLINK-30113
   
   
   ## Brief change log
   
   - sort the operator states by offsets before restoring.
   
   
   ## Verifying this change
   still working on tests
   
   ## Does this pull request potentially affect one of the following parts:
   
 - Dependencies (does it add or upgrade a dependency):  no
 - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: no
 - The serializers: no
 - The runtime per-record code paths (performance sensitive): no
 - Anything that affects deployment or recovery: yes
 - The S3 file system connector: no
   ## Documentation
   
 - Does this pull request introduce a new feature?  no
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[jira] [Created] (FLINK-33863) Compressed Operator state restore failed

2023-12-15 Thread Ruibin Xing (Jira)

Ruibin Xing created FLINK-33863:
---

Summary: Compressed Operator state restore failed
Key: FLINK-33863
URL: https://issues.apache.org/jira/browse/FLINK-33863
Project: Flink
Issue Type: Bug
Components: Runtime / State Backends
Affects Versions: 1.18.0
Reporter: Ruibin Xing

We encountered an issue when using Flink 1.18.0. Our job enabled Snapshot
Compression and used multiple Operator States in an operator. When recovering
Operator State from a Savepoint, the following error occurred:
"org.xerial.snappy.SnappyFramedInputStream: encountered EOF while reading
stream header."

After researching, I believe the error is due to Flink 1.18.0's support for
Snapshot Compression on Operator State (see
https://issues.apache.org/jira/browse/FLINK-30113 ). When writing a Savepoint,
SnappyFramedInputStream adds a header to the beginning of the data. When
recovering Operator State from a Savepoint, SnappyFramedInputStream verifies
the header from the beginning of the data.

Currently, when recovering Operator State with Snapshot Compression enabled,
the logic is as follows:
For each OperatorStateHandle:
1. Verify if the current Savepoint stream's offset is the Snappy header.
2. Seek to the state's start offset.
3. Read the state's data and finally seek to the state's end offset.
(See:
https://github.com/apache/flink/blob/ef2b626d67147797e992ec3b338bafdb4e5ab1c7/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java#L172
)

Furthermore, when there are multiple Operator States, they are not sorted
according to the Operator State's offset. Therefore, if the Operator States are
out of order and the final offset is recovered first, the Savepoint stream will
be seeked to the end, resulting in an EOF error.

I propose a solution: sort the OperatorStateHandle by offset and then recover
the Operator State in order. After testing, this approach resolves the issue.

1 2 >

1 - 100 of 153 matches

Mail list logo