feat(sessions): Add RedisSessionService implementation

[chenhr-1230]

Description of Change:

This PR introduces RedisSessionService, a new production-ready implementation of the BaseSessionService interface that uses Redis as a distributed, persistent session store.

Why is this change necessary?

Currently, the sessions module provides:

1. InMemorySessionService: Suitable for testing and single-node development, but is not persistent and does not support distributed environments.
2. VertexAiSessionService: A production-ready solution, but it couples the application to the Google Cloud Vertex AI platform.

This RedisSessionService fills a critical gap by providing a high-performance, distributed, and platform-agnostic solution that can be used by any developer deploying the ADK in a distributed environment (e.g., Kubernetes, on-prem) with a Redis backend.

Key Features & Design Decisions:

• Concurrency Handling: The implementation resolves concurrency races (e.g., two Runners or processes updating the same session simultaneously) by using Redis optimistic locking (WATCH/MULTI/EXEC). This is implemented in the appendEvent method, which atomically reads, modifies, and writes the session JSON.
• Data Model: Stores the entire Session object (including state and the events list) as a single JSON string per key in Redis. This is a clean K-V model.
• Dependencies:
  • Adds redis.clients:jedis as an <optional>true</optional> dependency in pom.xml, ensuring users who only use InMemory are not forced to download it.
  • Adds org.mockito:mockito-junit-jupiter as a <scope>test</scope> dependency to support the new unit tests.

Testing:

• Added RedisSessionServiceTest.java with full unit test coverage for all public methods.
• Tests utilize Mockito to mock the JedisPool and Jedis clients.
• Crucially, the optimistic locking logic is fully tested with three specific scenarios:
  1. appendEvent_shouldSucceedOnFirstTry (no conflict)
  2. appendEvent_shouldRetryOnConflictAndThenSucceed (simulated conflict and successful retry)
  3. appendEvent_shouldFailAfterMaxRetries (simulated persistent conflict)
  4. appendEvent_shouldUpdateStateFromStateDelta (verifies stateDelta is correctly merged)

All 8 tests pass successfully (exit code 0).

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[gemini-code-assist]

Summary of Changes

Hello @chenhr-1230, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a robust RedisSessionService to the ADK's sessions module, providing a production-ready, distributed, and platform-agnostic solution for managing user sessions. By utilizing Redis as a persistent store, it enables applications to scale and maintain session state across multiple instances. A key design aspect is the implementation of optimistic locking to ensure data integrity during concurrent session modifications, addressing a common challenge in distributed systems. The changes are thoroughly tested with dedicated unit tests covering various concurrency scenarios.

Highlights

• New RedisSessionService: Introduced a new RedisSessionService implementation for distributed and persistent session storage, leveraging Redis as a backend. This fills a critical gap by providing a high-performance, platform-agnostic solution for distributed environments.
• Optimistic Locking for Concurrency: The appendEvent method implements Redis optimistic locking using WATCH/MULTI/EXEC to handle concurrent updates to sessions, ensuring data consistency in a distributed setting.
• Session Data Model: The entire Session object, including its state and event list, is stored as a single JSON string per key in Redis, following a clean key-value model.
• Dependency Management: Added redis.clients:jedis as an optional dependency to pom.xml, allowing users to include it only if they utilize the Redis service. org.mockito:mockito-junit-jupiter was added as a test-scoped dependency.
• Comprehensive Unit Testing: New unit tests in RedisSessionServiceTest.java provide full coverage, specifically validating the optimistic locking logic with scenarios for successful operations, retries on conflict, and failure after max retries, along with state delta merging.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This is a great addition, providing a much-needed platform-agnostic, distributed session service using Redis. The implementation of optimistic locking with WATCH/MULTI/EXEC to handle concurrency is well-done, and the unit tests covering the conflict and retry scenarios are excellent. I have a few suggestions to improve the implementation further:

• There's a critical issue in appendEvent where it doesn't handle the removal of keys from the session state, which diverges from the behavior defined in BaseSessionService.
• The reactive stream design in getSession could be more idiomatic for handling "not found" cases.
• There are a couple of "magic numbers" for retry counts and scan sizes that would be better as named constants for improved readability and maintainability.

Please see my detailed comments in the code. Overall, this is a solid piece of work.

[gemini-code-assist]

The current implementation of applying stateDelta uses putAll, which does not handle the removal of keys from the session state, nor does it ignore temporary keys. The BaseSessionService interface's default appendEvent method includes logic to check for a special State.REMOVED value to delete a key and to ignore keys prefixed with State.TEMP_PREFIX. This implementation should replicate that logic to ensure consistent behavior. Without this, it's impossible to remove a key from the session state or correctly handle temporary state values.

                            ConcurrentMap<String, Object> sessionState = currentSession.state();
                            stateDelta.forEach(
                                (key, value) -> {
                                  if (!key.startsWith(State.TEMP_PREFIX)) {
                                    if (value == State.REMOVED) {
                                      sessionState.remove(key);
                                    } else {
                                      sessionState.put(key, value);
                                    }
                                  }
                                });

[gemini-code-assist]

The value 5 for the maximum number of retries is a "magic number". It's better to define it as a named constant to improve readability and make it easier to change.

You should also define the constant at the top of the class: private static final int DEFAULT_MAX_RETRIES = 5;

Suggested change

	this.maxRetries = 5; // 设置最大并发重试次数
	this.maxRetries = DEFAULT_MAX_RETRIES; // 设置最大并发重试次数

[gemini-code-assist]

The current implementation uses .switchIfEmpty(Maybe.error(new SessionNotFoundException(sessionId))) to handle a missing session. In reactive programming with RxJava, it's more idiomatic for a Maybe stream to simply complete without emitting an item (i.e., call onComplete) to signify "not found". Emitting an error is typically reserved for unexpected failures, not for an expected "not found" case. This forces consumers to handle a normal "not found" case in their onError block, which can be counter-intuitive. Consider removing this line to align with standard reactive patterns. The caller can then use .doOnComplete() or .switchIfEmpty() on their end if they need to treat "not found" as an error.

[gemini-code-assist]

The value 100 for the SCAN count is a magic number. It should be extracted into a private static final constant with a descriptive name, like DEFAULT_SCAN_COUNT. This makes the code more readable and easier to maintain.

You should also define the constant at the top of the class: private static final int DEFAULT_SCAN_COUNT = 100;

Suggested change

	ScanParams scanParams = new ScanParams().match(matchPattern).count(100);
	ScanParams scanParams = new ScanParams().match(matchPattern).count(DEFAULT_SCAN_COUNT);