feat(extstore): implement core extstore logic that uses payload visitor to batch and store/retrieve payloads. - #2976
Open
cconstable wants to merge 4 commits into
Open
feat(extstore): implement core extstore logic that uses payload visitor to batch and store/retrieve payloads.#2976cconstable wants to merge 4 commits into
cconstable wants to merge 4 commits into
Conversation
…or to batch and store/retrieve payloads.
cconstable
commented
Jul 29, 2026
|
|
||
| private CompletableFuture<List<IndexedValue<Payload>>> runStoreDrivers( | ||
| Map<String, Batch<Payload>> batches, @Nullable StorageDriverTargetInfo target) { | ||
| return TaskScope.withScope( |
Contributor
Author
There was a problem hiding this comment.
The previous logic here needed to manually account for external cancellation and handle sibling cancellation. The TaskScope handles that automatically. The scope.awaitAll() at the end has fail-fast semantics so it will immediately cancel other siblings on first error.
There was a problem hiding this comment.
Pull request overview
This PR introduces the core “extstore” plumbing in the Java SDK by adding internal converters that traverse proto messages, batch payloads by driver, and offload/retrieve large payloads via external storage drivers, with structured cancellation support.
Changes:
- Added internal payload converters and reference encoding/decoding to offload payloads to external storage and restore them on retrieval.
- Exposed payload-visitor utilities to support traversing nested proto payload fields and applying conversions with bounded concurrency.
- Extended storage driver contexts with cancellation tokens and added tests covering batching, thresholds, ordering, and cancellation behavior.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverStoreContext.java | Adds cancellation token to store context for aborting in-flight driver calls. |
| temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverRetrieveContext.java | Adds cancellation token to retrieve context for aborting in-flight driver calls. |
| temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriver.java | Documents cancellation token usage for driver implementers. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitors.java | Makes payload visitor entrypoints public for cross-package internal reuse. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitorOptions.java | Makes visitor options public to configure traversal/concurrency/skips. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitor.java | Makes the payload visitor functional interface public. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverStoreContextImpl.java | Implements store context with target info + cancellation token wiring. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageReferences.java | Encodes/decodes external storage reference payloads. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStoragePayloadConverter.java | Core batching logic to store/retrieve payload lists with TaskScope cancellation. |
| temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageMessageConverter.java | Bridges payload-list conversion over arbitrary proto messages via PayloadVisitors. |
| temporal-sdk/src/test/java/io/temporal/payload/storage/ExternalStorageOptionsTest.java | Updates tests for new cancellation token requirement in store context. |
| temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageReferencesTest.java | Adds tests for reference round-tripping and validation behavior. |
| temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStoragePayloadConverterTest.java | Adds tests for batching, thresholds, multi-driver routing, and cancellation on failure. |
| temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageMessageConverterTest.java | Adds tests for message traversal, nested payload conversion, and skipping search attributes. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+29
to
+31
| static boolean isReference(@Nonnull Payload payload) { | ||
| return payload.getExternalPayloadsCount() > 0; | ||
| } |
Contributor
Author
There was a problem hiding this comment.
I think the correct thing to do it replace isReference() + fromReferencePayload() with a single tryParseReference(Payload) that checks both external_payloads > 0 and messageType == temporal.api.sdk.v1.ExternalStorageReference and returns null instead of throwing. Will update this soon.
cconstable
commented
Jul 29, 2026
Comment on lines
+29
to
+31
| static boolean isReference(@Nonnull Payload payload) { | ||
| return payload.getExternalPayloadsCount() > 0; | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Going to replace this. See thread in conversation tab.
cconstable
commented
Jul 29, 2026
Comment on lines
+61
to
+70
| static ParsedReference fromReferencePayload(@Nonnull Payload payload) { | ||
| ByteString messageType = payload.getMetadataMap().get(EncodingKeys.METADATA_MESSAGE_TYPE_KEY); | ||
| if (messageType == null || !REFERENCE_MESSAGE_TYPE.equals(messageType.toStringUtf8())) { | ||
| throw new IllegalArgumentException( | ||
| "Payload is not an external storage reference; expected messageType '" | ||
| + REFERENCE_MESSAGE_TYPE | ||
| + "' but was '" | ||
| + (messageType == null ? "" : messageType.toStringUtf8()) | ||
| + "'"); | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Going to replace this. See thread in conversations tab.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What was changed
internal/payload/storage/ExternalStorageMessageConverter.javato transform payloads in messages.internal/payload/storage/ExternalStoragePayloadConverter.javato handle the actual transformation.TaskScopefor handling parent-child and sibling task cancellation.Why?
Checklist