Introduce inflight record index cache for bucket assign operator

[cshuo]

Task Description

When record level index is used in flink stream writing, the index data is committed in the downstream operator, so inflight record index cache should be introduced for uncommitted index data.

What needs to be done:

• Introduce inflight entries cache(for uncommitted checkpoint)

Why this task is needed:

Task Type

Other

Related Issues

Parent feature issue: (if applicable ) #17697
Related issues:
NOTE: Use Relationships button to add parent/blocking issues after issue is created.