[FLINK-25260] Recovery fails when using changelog+s3+presto

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.15.0
• Component/s: Connectors / FileSystem, Runtime / State Backends
• Assignee: Unassigned
• Reporter: Roman Khachatryan
• Created: 11/Dec/21 22:03
• Updated: 27/Dec/21 10:02
• Resolved: 27/Dec/21 10:02

Description

Recovery succeeds if using local FS or hadoop S3 plugin, but fails with Presto:

Caused by: java.lang.IllegalStateException
    at org.apache.flink.util.Preconditions.checkState(Preconditions.java:177)
    at org.apache.flink.changelog.fs.StateChangeFormat$1.readChange(StateChangeFormat.java:138)
    at org.apache.flink.changelog.fs.StateChangeFormat$1.next(StateChangeFormat.java:129)
    at org.apache.flink.changelog.fs.StateChangeFormat$1.next(StateChangeFormat.java:98)
    at org.apache.flink.runtime.state.changelog.StateChangelogHandleStreamHandleReader$1.next(StateChangelogHandleStreamHandleReader.java:76)
    at org.apache.flink.runtime.state.changelog.StateChangelogHandleStreamHandleReader$1.next(StateChangelogHandleStreamHandleReader.java:61)
    at org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation.readBackendHandle(ChangelogBackendRestoreOperation.java:94)
    at org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation.restore(ChangelogBackendRestoreOperation.java:74)
    at org.apache.flink.state.changelog.ChangelogStateBackend.restore(ChangelogStateBackend.java:221)
    at org.apache.flink.state.changelog.ChangelogStateBackend.createKeyedStateBackend(ChangelogStateBackend.java:145)
    at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:329)
    at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
    at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
    ... 13 more

This is likely caused by some intermediate buffers not being flushed.

Issue Links

• is fixed by: FLINK-25446 Avoid sanity check on read bytes on DataInputStream#read(byte[])
• relates to: FLINK-21352 FLIP-158: Generalized incremental checkpoints
• relates to: FLINK-25144 Test Changelog State Backend manually