runtime: retry Redis operations against the current client

[nabrahma]

What type of PR is this?

/kind bug

What this PR does / why we need it:

keys() and scan_keys() passed self._client.keys and self._client.scan into _execute_with_retry. Python resolves those attributes at the call site, so the bound method addressed whichever client object existed before the retry loop ran. On a ConnectionError the loop clears _connected and reconnects, and connect() assigns a new redis.Redis to self._client, but the retry kept invoking the old bound method. Every attempt went back to the connection that had already failed, and both methods then returned an empty list rather than raising, which a caller cannot tell apart from no keys matching.

Resolving self._client inside a lambda makes each attempt address the current client. It also makes the _ensure_connected call inside the loop effective, since self._client.scan previously raised AttributeError when the client was None, either before the first connect() or after _cleanup_connection().

Driving the real RedisClient with a fake whose client object is replaced on each connect(), counting calls per client generation:

before   scan {gen0: 4, gen1: 0, gen2: 0, gen3: 0}   returned []
after    scan {gen0: 1, gen1: 1}                     returned ['key-1']

Generation 0 is the dead client. Before, it absorbed all four attempts and the reconnected clients were never called.

Adds python/kthena/tests/test_redis_client.py, which had no tests. All five cases fail on main and pass here.

Which issue(s) this PR fixes:

Fixes #1618

Special notes for your reviewer:

Scoped to the callable change only. @aeron-gh suggested keeping the connection pool cleanup separate, and that is the right call for a second reason: doing it as _cleanup_connection() inside _execute_with_retry introduces a race. connect() is guarded by _connection_lock and _cleanup_connection() is not, so with two concurrent operations one coroutine's cleanup can disconnect the pool the other just established. I have a failing test for that, so the pool fix wants to live inside connect() under the lock, in its own PR.

Does this PR introduce a user-facing change?

Fixed the runtime Redis client retrying against the failed connection instead of the reconnected one, which made keys() and scan_keys() return an empty list after a transient connection error.

[volcano-sh-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign hzxuzhonghu for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• python/OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[nabrahma]

@LiZhenCheng9527 PTAL!