data-conversion

179 notes in this subject, read out of this brain and free to use. This is page 1 of 3.

ExternalStorage layer

The ExternalStorage layer is one of the three layers in the Data Converter and is optional. You need to customize it when your application requires payload offloading.

.NET IPayloadCodec interface methods

A custom IPayloadCodec must implement two async methods: EncodeAsync(IReadOnlyCollection<Payload>) and DecodeAsync(IReadOnlyCollection<Payload>). Both methods return Task<IReadOnlyCollection<Payload>>. The EncodeAsync method should convert payloads for encryption or compression before sending to the server. The DecodeAsync method should convert encrypted payloads back to readable form. Both methods should use the 'encoding' metadata field to track the encoding type and should not mutate the existing payloads.

.NET set custom PayloadCodec on TemporalClient

When creating a TemporalClient, you can set a custom PayloadCodec by updating the DataConverter property in the connection options. The syntax is: `DataConverter = DataConverter.Default with { PayloadCodec = new EncryptionCodec() }`. This applies the custom codec to all payloads sent to and received from the Temporal Cluster.

Custom Payload Codec in .NET SDK

A developer adds client-side encryption of Payloads by providing a Custom Payload Codec to the Client. If you only need to add compression or encryption to the default encoding, you can override the default Data Converter to use a new IPayloadCodec instead of creating a complete custom Data Converter.

.NET custom encryption codec example

```csharp public class EncryptionCodec : IPayloadCodec { public Task<IReadOnlyCollection<Payload>> EncodeAsync(IReadOnlyCollection<Payload> payloads) => Task.FromResult<IReadOnlyCollection<Payload>>(payloads.Select(p => { return new Payload() { Metadata = { ["encoding"] = "binary/my-payload-encoding" }, Data = ByteString.CopyFrom(Encrypt(p.ToByteArray())), }; }).ToList()); public Task<IReadOnlyCollection<Payload>> DecodeAsync(IReadOnlyCollection<Payload> payloads) => Task.FromResult<IReadOnlyCollection<Payload>>(payloads.Select(p => { if (p.Metadata.GetValueOrDefault("encoding") != "binary/my-payload-encoding") { return p; } return Payload.Parser.ParseFrom(Decrypt(p.Data.ToByteArray())); }).ToList()); private byte[] Encrypt(byte[] data) => Encoding.ASCII.GetBytes(Convert.ToBase64String(data)); private byte[] Decrypt(byte[] data) => Convert.FromBase64String(Encoding.ASCII.GetString(data)); } ``` This example shows how to implement a custom IPayloadCodec for the .NET SDK that encrypts payloads using base64 encoding on both the encode and decode paths.

Handler serialization requirements

Message handler parameters and return values must be serializable. Prefer data classes to multiple input parameters, as they allow adding fields without changing the calling signature. Keep in mind that serialization and deserialization can fail with the default data converter if a new field does not have a default value.

Default type-specific Payload Converters in Go SDK

The Go SDK provides the following type-specific Payload Converters, listed in the order applied by the default Data Converter: NewNilPayloadConverter(), NewByteSlicePayloadConverter(), NewProtoJSONPayloadConverter(), NewProtoPayloadConverter(), NewJSONPayloadConverter().

Setting custom Data Converter in Client options

To set your custom Payload Converter, use NewCompositeDataConverter and set it as the Data Converter in the Client options.

CompositeDataConverter for custom Payload Converters

Use CompositeDataConverter to apply custom, type-specific Payload Converters in a specified order. NewCompositeDataConverter creates a new instance of CompositeDataConverter from an ordered list of type-specific Payload Converters.

Add custom type conversion to default Data Converter

To add custom type conversion while keeping the defaults, use: dataConverter := converter.NewCompositeDataConverter( converter.NewNilPayloadConverter(), converter.NewByteSlicePayloadConverter(), converter.NewProtoJSONPayloadConverter(), converter.NewProtoPayloadConverter(), YourCustomPayloadConverter(), converter.NewJSONPayloadConverter(), ). This places your custom converter just before the default JSON fall-through.

Payload Converter basic concept

Temporal SDKs provide a default Payload Converter that converts custom data types to Payload and back. Payload Converters can be customized independently of a Payload Codec.

Replace default Data Converter with custom NewCompositeDataConverter

To replace the default Data Converter with a custom NewCompositeDataConverter, use: dataConverter := converter.NewCompositeDataConverter(YourCustomPayloadConverter())

When CompositeDataConverter is necessary

Defining a new Composite Data Converter is not always necessary to implement custom data handling. You can override the default Converter with a custom Codec, but a Composite Data Converter may be necessary for complex Workflow logic.

Custom PayloadCodec implementation in Go

A custom PayloadCodec implementation requires two functions: Encode for encryption/compression logic and Decode for decryption/decompression logic. The PayloadCodec must be used in an instance of CodecDataConverter that wraps a Data Converter to do Payload conversions, applying the custom encoding and decoding to the converted Payloads.

Decoded Workflow results not persisted to Cluster

Data decoding may be performed by your application logic during your Workflows or Activities as necessary, but decoded Workflow results are never persisted back to the Temporal Cluster. Instead, they are stored encoded on the Cluster. To view output in the temporal workflow show CLI command or Web UI, you need to provide an additional parameter to decode the data.

Set custom DataConverter in client.Dial options

Set your custom PayloadCodec with an instance of DataConverter in the Dial client options when creating the client. Example: ```go c, err := client.Dial(client.Options{ // Set DataConverter here to ensure that Workflow inputs and results are // encoded as required. DataConverter: mycodecpackage.DataConverter, }) ```

PayloadCodec Encode and Decode function example

The following example shows how to implement custom Encode and Decode functions in a PayloadCodec: ```go // Codec implements converter.PayloadEncoder for snappy compression. type Codec struct{} // Encode implements converter.PayloadCodec.Encode. func (Codec) Encode(payloads []*commonpb.Payload) ([]*commonpb.Payload, error) { result := make([]*commonpb.Payload, len(payloads)) for i, p := range payloads { // Marshal proto origBytes, err := p.Marshal() if err != nil { return payloads, err } // Compress b := snappy.Encode(nil, origBytes) result[i] = &commonpb.Payload{ Metadata: map[string][]byte{converter.MetadataEncoding: []byte("binary/snappy")}, Data: b, } } return result, nil } // Decode implements converter.PayloadCodec.Decode. func (Codec) Decode(payloads []*commonpb.Payload) ([]*commonpb.Payload, error) { result := make([]*commonpb.Payload, len(payloads)) for i, p := range payloads { // Decode only if it's our encoding if string(p.Metadata[converter.MetadataEncoding]) != "binary/snappy" { result[i] = p continue } // Uncompress b, err := snappy.Decode(nil, p.Data) if err != nil { return payloads, err } // Unmarshal proto result[i] = &commonpb.Payload{} err = result[i].Unmarshal(b) if err != nil { return payloads, err } } return result, nil } ```

Create CodecDataConverter wrapping Data Converter

To create a custom PayloadCodec in Go, wrap an instance of a Data Converter with your custom PayloadCodec using NewCodecDataConverter. Example: ```go // Create an instance of Data Converter with your codec. var DataConverter = converter.NewCodecDataConverter( converter.GetDefaultDataConverter(), NewPayloadCodec(), ) // Create an instance of PayloadCodec. func NewPayloadCodec() converter.PayloadCodec { return &Codec{} } ```

Data encoding happens on client when passing input

Data encoding is performed by the client using the converters and codecs provided by Temporal or your custom implementation when passing input to the Temporal Cluster. For example, plain text input is usually serialized into a JSON object, and can then be compressed or encrypted.

ExternalStorage layer purpose and default

The ExternalStorage layer offloads large payloads to an external store. This layer is optional and has no default implementation (all payloads are stored in Event History by default). It only needs to be customized when the application needs to offload large payloads.

WorkflowClient with custom Payload Converter configuration

DefaultDataConverter ddc = DefaultDataConverter.newDefaultInstance() .withPayloadConverterOverrides(new YourCustomPayloadConverter()); WorkflowClientOptions workflowClientOptions = WorkflowClientOptions.newBuilder().setDataConverter(ddc).build();

JacksonJsonPayloadConverter custom implementation

To create a custom JacksonJsonPayloadConverter, instantiate an ObjectMapper, add custom logic to it, and pass it to the JacksonJsonPayloadConverter constructor.

Default supported types in Java

The default Data Converter in the Java SDK supports converting: null, byte arrays, Protobuf messages (encoded with proto3 JSON), Jackson JSON, and anything JSON-serializable.

Custom Payload Converter code example

public class YourCustomPayloadConverter implements PayloadConverter { @Override public String getEncodingType() { return "json/plain"; } @Override public Optional<Payload> toData(Object value) throws DataConverterException { // Add your convert-to logic here. } @Override public <T> T fromData(Payload content, Class<T> valueClass, Type valueType) throws DataConverterException { // Add your convert-from logic here. } }

Setting custom Payload Converter in WorkflowClient

Use DefaultDataConverter.newDefaultInstance().withPayloadConverterOverrides() to set a custom Payload Converter, then pass it to WorkflowClientOptions via setDataConverter(). This configuration applies to both the Worker process and workflow execution startup.

PayloadConverter interface methods

The PayloadConverter interface requires three methods: getEncodingType() which returns a string determining which default conversion behavior to override, toData(Object value) which converts an object to a Payload, and fromData(Payload content, Class<T> valueClass, Type valueType) which converts a Payload back to a typed object.

Custom PayloadConverter implementation example

Create a custom class implementing PayloadConverter interface with getEncodingType() returning a string like "json/plain", toData() containing convert-to logic, and fromData() containing convert-from logic.

Payload Converter purpose

Payload Converters serialize application objects into a Payload (a binary form with metadata) and deserialize them back. Temporal uses Payload Converters to transport data between components.

When to create custom Payload Converter

A custom Payload Converter is needed only if the default Data Converter does not support custom types used by the application. Implementing custom Payload conversion is optional.

PayloadCodec layer purpose

The PayloadCodec layer transforms encoded payloads through operations like encryption and compression. It is optional and only needed when your application requires encryption or compression of payloads.

ExternalStorage layer purpose

The ExternalStorage layer handles payload offloading for large data. It is optional and only needed when your application requires payload offloading.

Set custom PayloadCodec with CodecDataConverter in WorkflowClient

To set a custom PayloadCodec implementation with DefaultDataConverter in your WorkflowClient, use CodecDataConverter: ```java WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); WorkflowClient client = WorkflowClient.newInstance( service, WorkflowClientOptions.newBuilder() .setDataConverter( new CodecDataConverter( DefaultDataConverter.newDefaultInstance(), Collections.singletonList(new YourCustomPayloadCodec()))) .build()); ``` Use this configured client both in your Worker process and to start your Workflow Executions.

Create custom PayloadCodec implementation for client-side encryption

A Temporal developer adds client-side encryption of Payloads by providing a custom Payload Codec to its Client. Create a custom implementation of PayloadCodec and use it in CodecDataConverter to set a custom Data Converter. The Payload Codec performs byte-to-byte conversion and must be set with a Data Converter.

PayloadCodec encode and decode methods

Define custom encryption and compression logic in the encode method and decryption and decompression logic in the decode method of your PayloadCodec implementation.

Java PayloadCodec implementation example with AES/GCM encryption

Example implementation of PayloadCodec for Java SDK showing encryption and decryption of payloads: ```java class YourCustomPayloadCodec implements PayloadCodec { static final ByteString METADATA_ENCODING = ByteString.copyFrom("binary/encrypted", StandardCharsets.UTF_8); private static final String CIPHER = "AES/GCM/NoPadding"; static final String METADATA_ENCRYPTION_CIPHER_KEY = "encryption-cipher"; static final ByteString METADATA_ENCRYPTION_CIPHER = ByteString.copyFrom(CIPHER, StandardCharsets.UTF_8); static final String METADATA_ENCRYPTION_KEY_ID_KEY = "encryption-key-id"; private static final Charset UTF_8 = StandardCharsets.UTF_8; @NotNull @Override public List<Payload> encode(@NotNull List<Payload> payloads) { return payloads.stream().map(this::encodePayload).collect(Collectors.toList()); } @NotNull @Override public List<Payload> decode(@NotNull List<Payload> payloads) { return payloads.stream().map(this::decodePayload).collect(Collectors.toList()); } private Payload encodePayload(Payload payload) { String keyId = getKeyId(); SecretKey key = getKey(keyId); byte[] encryptedData; try { encryptedData = encrypt(payload.toByteArray(), key); } catch (Throwable e) { throw new DataConverterException(e); } return Payload.newBuilder() .putMetadata(EncodingKeys.METADATA_ENCODING_KEY, METADATA_ENCODING) .putMetadata(METADATA_ENCRYPTION_CIPHER_KEY, METADATA_ENCRYPTION_CIPHER) .putMetadata(METADATA_ENCRYPTION_KEY_ID_KEY, ByteString.copyFromUtf8(keyId)) .setData(ByteString.copyFrom(encryptedData)) .build(); } private Payload decodePayload(Payload payload) { if (METADATA_ENCODING.equals( payload.getMetadataOrDefault(EncodingKeys.METADATA_ENCODING_KEY, null))) { String keyId; try { keyId = payload.getMetadataOrThrow(METADATA_ENCRYPTION_KEY_ID_KEY).toString(UTF_8); } catch (Exception e) { throw new PayloadCodecException(e); } SecretKey key = getKey(keyId); byte[] plainData; Payload decryptedPayload; try { plainData = decrypt(payload.getData().toByteArray(), key); decryptedPayload = Payload.parseFrom(plainData); return decryptedPayload; } catch (Throwable e) { throw new PayloadCodecException(e); } } else { return payload; } } private String getKeyId() { return "test-key-test-key-test-key-test!"; } private SecretKey getKey(String keyId) { return new SecretKeySpec(keyId.getBytes(UTF_8), "AES"); } } ```

TypeScript plugin example with custom data converter

Example of registering a custom data converter in a TypeScript plugin: const codec: PayloadCodec = { encode(_payloads: Payload[]): Promise<Payload[]> { throw new Error(); }, decode(_payloads: Payload[]): Promise<Payload[]> { throw new Error(); }, }; const plugin = new SimplePlugin({ name: 'organization.PluginName', dataConverter: (converter: DataConverter | undefined) => ({ ...converter, payloadCodecs: [...(converter?.payloadCodecs ?? []), codec], }), });

DotNet plugin example with custom data converter

Example of registering a custom data converter in a DotNet plugin: private class Codec : IPayloadCodec { public Task<IReadOnlyCollection<Payload>> EncodeAsync(IReadOnlyCollection<Payload> payloads) => throw new NotImplementedException(); public Task<IReadOnlyCollection<Payload>> DecodeAsync(IReadOnlyCollection<Payload> payloads) => throw new NotImplementedException(); } SimplePlugin converterPlugin = new SimplePlugin( "organization.PluginName", new SimplePluginOptions() { DataConverterOption = new SimplePluginOptions.SimplePluginOption<DataConverter>( (converter) => converter with { PayloadCodec = new Codec() } ), });

Java plugin example with custom data converter

Example of registering a custom data converter in a Java plugin: SimplePlugin converterPlugin = SimplePlugin.newBuilder("organization.PluginName") .customizeDataConverter( existingConverter -> { // Customize the data converter // This example keeps the existing converter unchanged // In practice, you might wrap it with additional functionality return existingConverter; }) .build();

Ruby plugin example with custom data converter

Example of registering a custom data converter in a Ruby plugin: custom_converter = Temporalio::Converters::DataConverter.new( payload_converter: Temporalio::Converters::PayloadConverter.default ) plugin = Temporalio::SimplePlugin.new( name: 'organization.PluginName', data_converter: custom_converter )

Python plugin example with custom data converter

Example of registering a custom data converter in a Python plugin: def set_converter(converter: DataConverter | None) -> DataConverter: if converter is None or converter == DataConverter.default: return pydantic_data_converter # Should consider interactions with other plugins, # as this will override the data converter. # This may mean failing, warning, or something else return converter plugin = SimplePlugin("organization.PluginName", data_converter=set_converter)

Go plugin example with custom data converter

Example of registering a custom data converter in a Go plugin: func createConverterPlugin() (*temporal.SimplePlugin, error) { customConverter := converter.GetDefaultDataConverter() // Or your custom converter return temporal.NewSimplePlugin(temporal.SimplePluginOptions{ Name: "organization.PluginName", DataConverter: customConverter, }) }

Configure PayloadCodec on Data Converter

To use a PayloadCodec, add a data_converter parameter to Client.connect() options that overrides the default converter. The data_converter should use dataclasses.replace() on temporalio.converter.default() with payload_codec set to your PayloadCodec instance.

Payload encryption use case for codecs

The most common use case for Payload Codecs is encryption: encrypting payloads before they reach the Temporal Service so that sensitive data is never stored in plaintext.

Payload Codec example with snappy compression

This example shows a CompressionCodec that compresses payloads using Python's cramjam library with snappy compression: ```python import cramjam from temporalio.api.common.v1 import Payload from temporalio.converter import PayloadCodec class CompressionCodec(PayloadCodec): async def encode(self, payloads: Iterable[Payload]) -> List[Payload]: return [ Payload( metadata={ "encoding": b"binary/snappy", }, data=(bytes(cramjam.snappy.compress(p.SerializeToString()))), ) for p in payloads ] async def decode(self, payloads: Iterable[Payload]) -> List[Payload]: ret: List[Payload] = [] for p in payloads: if p.metadata.get("encoding", b"").decode() != "binary/snappy": ret.append(p) continue ret.append(Payload.FromString(bytes(cramjam.snappy.decompress(p.data)))) return ret ``` The encode() method compresses each payload and sets the encoding metadata. The decode() method checks the encoding metadata and decompresses payloads that were compressed with snappy, passing through other payloads unchanged.

PayloadCodec interface methods

A PayloadCodec must implement encode() and decode() methods. The encode() method transforms Payload bytes after serialization by the Payload Converter and before data is sent to the Temporal Service. The decode() method reverses the encode() logic. Both methods should loop through all of a Workflow's payloads, perform marshaling, compression, or encryption steps in order, and set an 'encoding' metadata field.

PayloadCodec runs outside workflow sandbox

Unlike Payload Converters, Payload Codecs run outside the Workflow sandbox, which means they can use non-deterministic operations and call external services.

Client.connect with PayloadCodec configuration example

This example shows how to configure a PayloadCodec when connecting a client: ```python from codec import CompressionCodec client = await Client.connect( "localhost:7233", data_converter=dataclasses.replace( temporalio.converter.default(), payload_codec=CompressionCodec() ), ) ```

Codec Server for Web UI and CLI payload decoding

A Codec Server is an HTTP server that runs a PayloadCodec remotely, allowing the Temporal Web UI and CLI to decode encrypted payloads for display.

Automatic payload handling with External Storage

All Workflows and Activities running on the Worker use the storage driver automatically without changes to your business logic. The driver uploads and downloads payloads concurrently and validates payload integrity on retrieve.

Custom StorageDriver implementation

A custom storage driver extends the StorageDriver abstract class and implements three methods: name() returns a unique string identifying the driver (changing this name after payloads are stored breaks retrieval); store() receives a list of Payload protobuf messages and returns one StorageDriverClaim per payload with key-value pairs for locating the payload later; retrieve() receives the claims and returns the original payloads. Convert payloads to bytes with payload.SerializeToString() when storing, and reconstruct them with payload.ParseFromString(data) when retrieving.

S3 Multi-Region Access Point for cross-region replication

To make S3-backed External Storage tolerant of regional failures, configure AWS with Cross-Region Replication (CRR) and an S3 Multi-Region Access Point (MRAP), then point the driver at the MRAP ARN instead of a bucket name. The only code change is passing the MRAP ARN as the bucket value: bucket="arn:aws:s3::123456789012:accesspoint/mfzwi23gnjvgw.mrap". aioboto3 (via botocore) automatically uses SigV4A signing when the bucket value is an MRAP ARN. Ensure your botocore version is recent enough to support SigV4A.

StorageDriverStoreContext provides target identity information

The context.target in the store() method provides identity information such as namespace, Workflow ID, or Activity ID depending on the operation. Consider structuring storage keys to include this information so that you can identify which Workflow owns each payload.

S3StorageDriver setup with aioboto3

Create an S3 client using aioboto3 and pass it to the S3StorageDriver. The driver uses standard AWS credentials from the environment (environment variables, IAM role, or AWS config file). Example: session = aioboto3.Session(profile_name=AWS_PROFILE, region_name=AWS_REGION) async with session.client("s3") as s3_client: driver = S3StorageDriver( client=new_aioboto3_client(s3_client), bucket="my-temporal-payloads", )

Configure ExternalStorage on DataConverter

Pass an ExternalStorage instance to your DataConverter and use the converter when creating your Client and Worker. Example: data_converter = dataclasses.replace( DataConverter.default, external_storage=ExternalStorage(drivers=[driver]), ) client = await Client.connect(**client_config, data_converter=data_converter) worker = Worker( client, task_queue="my-task-queue", workflows=[], activities=[], )

Configure custom storage driver on DataConverter

Pass an ExternalStorage instance with your custom driver to the DataConverter: data_converter = dataclasses.replace( DataConverter.default, external_storage=ExternalStorage( drivers=[LocalDiskStorageDriver()], ), )

External Storage default payload size threshold

By default, payloads of 256 KiB or larger are offloaded to external storage. This threshold can be adjusted with the payload_size_threshold parameter, or set to 0 to externalize all payloads regardless of size. Payloads smaller than the threshold stay inline in Event History.

External Storage with S3 - prerequisites

To use External Storage with Amazon S3, you need an Amazon S3 bucket with read and write access. You must also install the aioboto3 extra with: python -m pip install "temporalio[aioboto3]". Refer to lifecycle management to ensure payloads remain available for the entire lifetime of the Workflow.

Custom storage driver example using local disk

class LocalDiskStorageDriver(StorageDriver): def __init__(self, store_dir: str = "/tmp/temporal-payload-store") -> None: self._store_dir = store_dir def name(self) -> str: return "local-disk" async def store( self, context: StorageDriverStoreContext, payloads: Sequence[Payload], ) -> list[StorageDriverClaim]: os.makedirs(self._store_dir, exist_ok=True) prefix = self._store_dir target = context.target if isinstance(target, StorageDriverWorkflowInfo) and target.id: prefix = os.path.join(self._store_dir, target.namespace, target.id) os.makedirs(prefix, exist_ok=True) claims = [] for payload in payloads: key = f"{uuid.uuid4()}.bin" file_path = os.path.join(prefix, key) with open(file_path, "wb") as f: f.write(payload.SerializeToString()) claims.append(StorageDriverClaim(claim_data={"path": file_path})) return claims async def retrieve( self, context: StorageDriverRetrieveContext, claims: Sequence[StorageDriverClaim], ) -> list[Payload]: payloads = [] for claim in claims: file_path = claim.claim_data["path"] with open(file_path, "rb") as f: raw = f.read() payload = Payload() payload.ParseFromString(raw) payloads.append(payload) return payloads

Multiple storage drivers with driver selector

When registering multiple drivers, provide a driver_selector function that chooses which driver stores each payload. Any driver in the list that is not selected for storing is still available for retrieval, useful when migrating between storage backends. Return None from the selector to keep a specific payload inline in Event History. Example: preferred_driver = S3StorageDriver(client=s3_client, bucket="my-bucket") legacy_driver = LegacyStorageDriver() ExternalStorage( drivers=[preferred_driver, legacy_driver], driver_selector=lambda context, payload: preferred_driver, )

PayloadCodec interface in Ruby SDK

The Payload Codec needs to extend Temporalio::Converters::PayloadCodec and implement encode and decode methods. The encode method should convert payloads into new payloads by setting the 'encoding' metadata field. The decode method should convert encoded payloads back, also using the 'encoding' metadata field. Do not mutate the existing payloads.

Give your agent this brain