...

We would apply both parameters in GlobalStateManagerImpl to guard against TimeoutException for KafkaCosumer#partitionsFor and KafkaConsumer#endOffsets.

Furthermore, we can apply both parameters for retrying getting state locks (cf. StateDirectory) that is hard coded with five retires at the moment.

To align with current design, we would also pass the new parameter to producer and consumer if there is no "producer." or "consumer." prefix. Thus, users can still configure different values for consumer/producer if needed.

...

• This change is backwards compatible as we only add new parameters and also keep the current behavior with default parameter settings.State locking retry would be disabled by default (currently hard coded of 5), but this should be ok.

Test Plan

We add tests with mocking the global consumer client to throw TimeoutException to check if retry parameters are respected.

...