[FLINK-37158][tests] Introduce ForSt to existing ITCases

[fredia]

What is the purpose of the change

Introduce ForSt to existing IT Cases.

Brief change log

• Enable Async State in OneInputTransformation
• Introduce forst to AutoRescalingITCase, EventTimeWindowCheckpointingITCase, KeyedStateCheckpointingITCase
  , RescaleCheckpointManuallyITCase.

Verifying this change

This change is a trivial rework , which is already covered by existing 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: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[flinkbot]

CI report:

• 11a8ec1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[Zakelly]

LGTM

[Zakelly]

@flinkbot run azure

[Zakelly]

@flinkbot run azure

[Zakelly]

@flinkbot run azure

[Zakelly]

@fredia The stacktrace of CI failure:

Caused by: java.io.IOException: Error while opening ForSt instance.
	at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:97)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.restoreTempDBInstance(ForStIncrementalRestoreOperation.java:950)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.copyToBaseDBUsingTempDBs(ForStIncrementalRestoreOperation.java:593)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.mergeStateHandlesWithCopyFromTemporaryInstance(ForStIncrementalRestoreOperation.java:634)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.restoreFromMultipleStateHandles(ForStIncrementalRestoreOperation.java:353)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.innerRestore(ForStIncrementalRestoreOperation.java:270)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.lambda$restore$1(ForStIncrementalRestoreOperation.java:227)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.runAndReportDuration(ForStIncrementalRestoreOperation.java:427)
	at org.apache.flink.state.forst.restore.ForStIncrementalRestoreOperation.restore(ForStIncrementalRestoreOperation.java:227)
	at org.apache.flink.state.forst.sync.ForStSyncKeyedStateBackendBuilder.build(ForStSyncKeyedStateBackendBuilder.java:313)
	... 19 more
Caused by: org.forstdb.RocksDBException: While open a file for appending: /tmp/junit3413624162248896720/junit2347279978830062410/minicluster_1c75ad2fbab9c1daad37de9e2c77cea7/tm_1/tmp/job_94ae7dea42153ce4c7fbee25d1db7630_op_StreamFlatMap_20ba6b65f97481d5570070de90e4e791__2_2__uuid_34ea6969-141f-495d-a048-a601f4350d35/db/tmp_junit3413624162248896720_junit2347279978830062410_minicluster_1c75ad2fbab9c1daad37de9e2c77cea7_tm_1_tmp_job_94ae7dea42153ce4c7fbee25d1db7630_op_StreamFlatMap_20ba6b65f97481d5570070de90e4e791__2_2__uuid_34ea6969-141f-495d-a048-a601f4350d35_3aa732f2-b8c_LOG: File name too long
	at org.forstdb.RocksDB.open(Native Method)
	at org.forstdb.RocksDB.open(RocksDB.java:317)
	at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:85)
	... 28 more

Seems the DB Log path is too long. Need a redirection? Will this a problem for user jobs?

[davidradl]

Reviewed by Chi on 23/01/2025 Need a committer to review. CI failing - in the area of the fix.

[fredia]

@flinkbot run azure

[Zakelly]

Seems after rebasing, the tests are fine. LGTM