Token Store Migration

Axon Framework 5 adds a mask column to the token store table used by streaming event processors. This column is required: without it, the framework cannot start. If you are migrating from Axon Framework 4, the existing schema must be updated and the column must be populated with correct values.

Alongside this schema change, the tracking token classes moved to a new package, so the class names stored in existing token data also need attention. See Token class names for what that means and when you need to act.

Understanding the mask column

Streaming event processors divide the event stream into segments. Each row in the token store represents one segment for a given processor. In Axon Framework 5, a mask column is stored alongside the segment column.

The mask is a bitmask that, together with the segment ID, determines which events that segment processes. It is derived from how many times the root segment has been split and changes whenever the processor scales up or down.

For a processor with four equal-size segments, the token store contains:

processorName segment mask

inventory-projection

0

3

inventory-projection

1

3

inventory-projection

2

3

inventory-projection

3

3

For a processor with two segments:

processorName segment mask

inventory-projection

0

1

inventory-projection

1

1

The non-trivial case is when the number of segments is not a power of two. A processor that started with two segments and then had one of them split produces three segments:

processorName segment mask

inventory-projection

0

3

inventory-projection

1

1

inventory-projection

2

3

Segments 0 and 2 (the split pair) have mask 3, while segment 1 (which was never split) retains mask 1. The correct mask for each segment can be determined unambiguously from the complete set of segment IDs for that processor, but not from any individual segment ID in isolation. This is what makes the migration non-trivial: you must read all segment IDs for a processor together and compute their masks as a group.

Choosing a migration strategy

Three strategies are available. Choose based on your constraints:

Strategy Description Downtime

Scale to single segment

Scale all processors to one segment before migrating. The mask is trivially 0.

Brief outage

Reset processors to latest

Delete all token entries and let Axon Framework 5 create them fresh at the latest position.

Brief outage

Programmatic mask migration

Compute and set the correct mask for every existing token entry.

Zero downtime possible

Option 1: Scale to a single segment

This is the simplest strategy. Before the schema change, scale every processor down to a single segment. The single remaining segment is always the root segment (segment ID 0, mask 0), where both segment ID and mask are 0. The SQL migration is a single ALTER TABLE.

Steps:

  1. Scale all event processors to a single segment by repeatedly merging segments until only segment 0 remains. See Splitting and merging segments for how to trigger a merge.

  2. Shut down the application.

  3. Apply the schema change:

    ALTER TABLE token_entry ADD COLUMN mask INTEGER NOT NULL DEFAULT 0;

  4. Upgrade to Axon Framework 5 and restart.

  5. Scale the processors back up to the desired number of segments.

After merging down to a single segment, every processor has exactly one row with segment = 0. The root segment always has both segment ID and mask equal to 0, so the default of 0 is correct.

Option 2: Reset processors to latest

If scaling processors down first is impractical, you can delete the existing token entries and let Axon Framework 5 reinitialize them at the latest position in the event stream. No mask computation is needed because the framework creates the new entries with correct mask values.

For this to be safe, the processors must be fully caught up before the restart, so that no events are missed when they reinitialize at the latest position.

Steps:

  1. Halt all writes to the application so that no new events are produced.

  2. Wait for all event processors to catch up to the latest position in the event stream.

  3. Shut down the application.

  4. Add the mask column and delete all existing token entries:

    ALTER TABLE token_entry ADD COLUMN mask INTEGER NOT NULL DEFAULT 0;
DELETE FROM token_entry;

  5. Configure the processors to start at the latest position when no token entry exists, using LatestTrackingToken. Refer to Streaming Event Processor for the initial token configuration.

  6. Upgrade to Axon Framework 5 and restart. The processors find no existing tokens, initialize at the latest position, and create fresh entries with the correct mask values.

  7. Re-enable writes.

Events that arrive between step 1 (halting writes) and step 7 (re-enabling writes) are not processed by the reset processors, because they initialize at the latest position that existed at startup time. After writes are re-enabled, the processors continue from that point forward.

Option 3: Programmatic mask migration

This option computes the correct mask for every existing token entry and updates the table in place. It can run while the application is offline or be integrated into a schema migration tool such as Flyway or Liquibase. No scaling or token deletion is required.

The mask computation mirrors the split logic of org.axonframework.messaging.eventhandling.processing.streaming.segmenting.Segment. Given the full set of segment IDs for a processor, the algorithm reconstructs the segment hierarchy by simulating splits from the root and identifies the mask for each leaf segment.

Add the column

First, add the mask column as nullable so existing rows are not immediately rejected:

ALTER TABLE token_entry ADD COLUMN mask INTEGER;

Flyway Java migration

Add a versioned Flyway Java migration that reads the segment IDs, computes the masks, and writes them back:

import org.flywaydb.core.api.migration.BaseJavaMigration;
import org.flywaydb.core.api.migration.Context;

import java.sql.*;
import java.util.*;

public class V2__PopulateTokenStoreMask extends BaseJavaMigration {

    @Override
    public void migrate(Context context) throws Exception {
        Connection conn = context.getConnection();
        Map<String, List<Integer>> segmentsByProcessor = readSegments(conn);
        updateMasks(conn, segmentsByProcessor);
    }

    private Map<String, List<Integer>> readSegments(Connection conn) throws SQLException {
        Map<String, List<Integer>> result = new LinkedHashMap<>();
        try (PreparedStatement stmt = conn.prepareStatement(
                "SELECT processorName, segment FROM token_entry ORDER BY processorName, segment");
             ResultSet rs = stmt.executeQuery()) {
            while (rs.next()) {
                result.computeIfAbsent(rs.getString("processorName"), k -> new ArrayList<>())
                      .add(rs.getInt("segment"));
            }
        }
        return result;
    }

    private void updateMasks(Connection conn,
                             Map<String, List<Integer>> segmentsByProcessor) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(
                "UPDATE token_entry SET mask = ? WHERE processorName = ? AND segment = ?")) {
            for (Map.Entry<String, List<Integer>> entry : segmentsByProcessor.entrySet()) {
                String processorName = entry.getKey();
                int[] ids = entry.getValue().stream().mapToInt(Integer::intValue).toArray();
                for (SegmentInfo seg : computeSegments(ids)) {
                    stmt.setInt(1, seg.mask());
                    stmt.setString(2, processorName);
                    stmt.setInt(3, seg.segmentId());
                    stmt.addBatch();
                }
            }
            stmt.executeBatch();
        }
    }

    private static List<SegmentInfo> computeSegments(int[] segmentIds) {
        if (segmentIds.length == 0) {
            return List.of();
        }
        List<Integer> ids = Arrays.stream(segmentIds).boxed().toList();
        Set<SegmentInfo> result = new HashSet<>();
        computeSegments(new SegmentInfo(0, 0), ids, result);
        return result.stream()
                     .sorted(Comparator.comparingInt(SegmentInfo::segmentId))
                     .toList();
    }

    private static void computeSegments(SegmentInfo current, List<Integer> segmentIds,
                                        Set<SegmentInfo> result) {
        int newMask = (current.mask() << 1) | 1;
        int sibling = current.segmentId() + (current.mask() == 0 ? 1 : newMask ^ current.mask());
        if (segmentIds.contains(sibling)) {
            computeSegments(new SegmentInfo(current.segmentId(), newMask), segmentIds, result);
            computeSegments(new SegmentInfo(sibling, newMask), segmentIds, result);
        } else {
            result.add(current);
        }
    }

    private record SegmentInfo(int segmentId, int mask) {}
}

Apply the NOT NULL constraint

Once the migration has run successfully, enforce the constraint:

-- PostgreSQL
ALTER TABLE token_entry ALTER COLUMN mask SET NOT NULL;

-- MySQL / MariaDB
ALTER TABLE token_entry MODIFY COLUMN mask INTEGER NOT NULL;

-- SQL Server
ALTER TABLE token_entry ALTER COLUMN mask INTEGER NOT NULL;

Liquibase

For Liquibase, extract the computeSegments logic above into a helper class and invoke it from a CustomTaskChange implementation. Use a <changeSet> to add the nullable column first, then call the custom change to populate it, and finally apply the NOT NULL constraint in a third changeset.

Table and column names

The SQL examples above use the default JDBC token store column names. Adjust them to match your actual configuration:

Store type Default table name Column names

JPA with Spring Boot (SpringPhysicalNamingStrategy)

token_entry

processor_name, segment, mask

JDBC (TokenSchema defaults)

TokenEntry

processorName, segment, mask

JPA column names depend on the Hibernate physical naming strategy configured in your application. Spring Boot’s default SpringPhysicalNamingStrategy converts camelCase field names to snake_case. Using PhysicalNamingStrategyStandardImpl keeps them as-is. Check your persistence configuration to confirm the exact column names in your database before running any migration script.

Token class names

Between Axon Framework 4 and Axon Framework 5 the tracking token classes moved to a new package. A token store written by Axon Framework 4 records the old class names, for example org.axonframework.eventhandling.GlobalSequenceTrackingToken. The name appears in the token type column, and for tokens that wrap other tokens it also appears inside the serialized token. Whether you need to act depends on how those tokens were serialized.

Jackson (JSON) stores

If your Axon Framework 4 application serialized its tokens with Jackson, no action is required, as long as Axon Framework 5 reads them with a Jackson-based converter (the default). This applies to both the Jackson 3 and the Jackson 2 converter. Axon Framework 5 recognizes the old class names when reading a token and maps them to the new classes. The first time each token is updated it is stored under the new name, so the store migrates itself as the processors run.

XStream stores

Axon Framework 5 does not include XStream, so it cannot read tokens serialized with the removed XStreamSerializer. See the Serializer Migration for the background on that removal. A token only records the position a streaming event processor has reached, so these tokens can be regenerated. The simplest option is to reset the processors to latest and let them rebuild.

If a full replay is not acceptable, re-serialize the tokens to JSON while still on Axon Framework 4, before upgrading. Run the following for each token entry, reading the stored token with the serializer your Axon Framework 4 application already uses and writing it back with Jackson:

Serializer xStreamSerializer = /* the serializer your Axon 4 application is configured with */;
Serializer jacksonSerializer = JacksonSerializer.defaultSerializer();

// tokenColumn and tokenTypeColumn hold the current values of the token entry
SerializedObject<byte[]> stored =
        new SimpleSerializedObject<>(tokenColumn, byte[].class, tokenTypeColumn, null);
TrackingToken token = xStreamSerializer.deserialize(stored);

SerializedObject<byte[]> jsonToken = jacksonSerializer.serialize(token, byte[].class);
// store jsonToken.getData() back in the token column

Only the token column changes to JSON. The token type column can keep the old class name, because Axon Framework 5 maps it on read. After this, upgrade and apply the mask migration above.

Replay tokens

Replay tokens are not migrated automatically, because their format changed between versions beyond the class name. Axon Framework 4 named the reset context field context, and Axon Framework 5 renamed it to resetContext and changed its type. A replay token only exists while a processor is replaying, so complete any replay in progress before upgrading, or reset the affected processors.

Multi-source tokens

Reading from more than one event source, and its MultiSourceTrackingToken, is part of the Axoniq Framework rather than Axon Framework 5 itself. When you use Axoniq Framework, these tokens are recognized and read after upgrading, in the same way as the tokens above. The exception is a token captured while one of its sources was replaying, because it contains a replay token, which cannot be migrated for the reason described under Replay tokens. Reset those processors instead.

On Axon Framework 5 without the Axoniq Framework there is no multi-source support, so reset the affected processors to rebuild their position, as described in reset the processors to latest.