fix(reader): implement retry logic for opening Redis reader

[sauravkumarrr]

Summary

• Add retry logic (3 attempts, 1s backoff) around reader.open() in RedisKeysSourceTask.start() to handle transient ConcurrentModificationException from spring-batch-redis's JobUtils.hsqldbDataSource() — the static HashMap.computeIfAbsent is not thread-safe on Java 9+
• Replace RetriableException with ConnectException after exhausting retries, since RetriableException has no effect during task startup (Connect treats it as unrecoverable and kills the task)
• Wrap reader.close() and client.shutdown() in try-catch in stop() to prevent zombie connectors when cleanup fails on partially-initialized state

Context

In production, a single keys source connector (lcc-99wr75) entered a 2+ week death spiral:

1. ConcurrentModificationException in HashMap.computeIfAbsent during reader.open() → task killed as "unrecoverable"
2. Restart attempts fail because the task can't be gracefully stopped (reader/client cleanup throws)
3. Connector becomes unresponsive zombie, consuming resources indefinitely

The root cause is in spring-batch-redis-infrastructure (tested 4.6.2 and 4.8.0) — JobUtils.dataSources is a plain HashMap instead of ConcurrentHashMap. This fix works around the upstream bug.

---

Note

Medium Risk
Touches Kafka Connect task startup/shutdown paths and introduces blocking retry sleeps, which can affect connector availability and thread usage if misbehaving. Scope is limited to the Redis keys source task and adds bounded retries with clearer failure behavior.

Overview
Improves resilience of the Redis keys source task during startup and shutdown. RedisKeysSourceTask.start() now opens the RedisItemReader via a new openReaderWithRetry() helper that retries reader.open() up to 3 times with a 1s delay, logs failures, closes partially-open readers between attempts, and ultimately fails with ConnectException (replacing the prior RetriableException startup behavior).

Hardens cleanup to prevent stuck/zombie tasks. stop() now wraps reader.close(), client.shutdown(), and client.getResources().shutdown() in try/catch with warnings so shutdown proceeds even if some resources fail to close.

^{Reviewed by Cursor Bugbot for commit b2265f5. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 40f436d. Configure here.}