KAFKA-14412: Add statestore.uncommitted.max.bytes

[nicktelford]

This configuration controls the maximum amount of local state store data to buffer in transaction buffers.

When the total uncommitted data, across all stores and all threads, exceeds this value, a Task commit will be triggered, even if commit.interval.ms has not yet elapsed. This prevents application from hitting OutOfMemory errors when the commit.interval.ms is set to a high value, and transactional state stores are in-use.

StateStores that support transactions can implement approximateNumUncommittedBytes to track the size of the data not yet committed to the store.

By default, approximateUncommitted returns 0, which means that all data is immediately committed to the state store.

[nicktelford]

Note: this is part of KIP-892 Transactional StateStores. It introduces a new config option that will have no effect until the rest of KIP-892 lands at a later date.

@cadonna @ableegoldman

[nicktelford]

@cadonna @ableegoldman Is there any more we need to do with this PR before we can merge it?

[github-actions[bot]]

This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch)

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.