You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pa...@apache.org on 2026/10/05 12:09:08 UTC
(flink) branch master updated: [FLINK-40297][runtime] Route TTL-aware value migration through the migrate hook
This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 91bbaa59dda [FLINK-40297][runtime] Route TTL-aware value migration through the migrate hook
91bbaa59dda is described below
commit 91bbaa59dda19d7d4b68c376fc4736fa719cbef9
Author: Weiqing Yang <ya...@gmail.com>
AuthorDate: Sun Aug 2 19:57:36 2026 -0700
[FLINK-40297][runtime] Route TTL-aware value migration through the migrate hook
TtlAwareSerializer.migrateValueFromPriorSerializer is the single entry point
through which the RocksDB state backend migrates state values on restore:
AbstractRocksDBState, RocksDBListState and RocksDBMapState all call it after
unwrapping the state shape they own. It deserialized with the prior serializer
and re-serialized with the new one, leaving a serializer no opportunity to adapt
the value in between.
Route it through TypeSerializerSnapshot.migrate: unwrap the prior value to its
bare, non-TTL form, migrate it, then re-wrap when this serializer is TTL-enabled,
preserving the prior timestamp when the prior value carried one. Behavior is
unchanged, because no serializer overrides the hook yet and its default returns
the value unchanged.
The hook receives the persisted prior snapshot rather than one re-derived by
calling snapshotConfiguration() on the restored prior serializer. That round trip
is lossy: PojoSerializerSnapshot substitutes a synthetic name for a field that no
longer exists on the class, so a migrate override reconciling fields by name
would see a fabricated schema. The backend already holds the persisted snapshot
above the migration loop, and each caller now descends it alongside its
serializer.
The descent unwraps the TtlAware decorator first. Registering a new serializer
mutates the previous snapshot's nested snapshots in place, so a list or map
state's persisted element or value snapshot is a TtlAwareSerializerSnapshot
rather than the snapshot the checkpoint wrote. A snapshot of an unexpected type
now fails rather than falling back to a re-derived one; only an absent snapshot
falls back.
Generated-by: Claude Code (Opus 5)
---
.../common/typeutils/TypeSerializerSnapshot.java | 15 +-
.../runtime/state/ttl/TtlAwareSerializer.java | 133 ++++++-
.../runtime/state/StateSerializerProviderTest.java | 41 +++
.../runtime/state/ttl/TtlAwareSerializerTest.java | 398 +++++++++++++++++++++
.../state/ttl/TtlAwareSerializerUpgradeTest.java | 1 +
.../flink/state/rocksdb/AbstractRocksDBState.java | 5 +
.../state/rocksdb/RocksDBKeyedStateBackend.java | 8 +-
.../flink/state/rocksdb/RocksDBListState.java | 21 ++
.../flink/state/rocksdb/RocksDBMapState.java | 21 ++
9 files changed, 625 insertions(+), 18 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
index e4788736a68..45b56d9fa10 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
@@ -150,10 +150,19 @@ public interface TypeSerializerSnapshot<T> {
* <p>The migration is not applied recursively to nested serializers. The snapshot of a
* composite type returns its value unchanged unless it overrides this method to decompose the
* value and migrate each part, so a caller that needs a nested value migrated must reach the
- * nested snapshot itself.
+ * nested snapshot itself. An implementation that does so should not assume that the old and the
+ * new snapshot expose nested snapshots of the same type: restoring may have replaced those of
+ * the old snapshot with decorators, so nested snapshots are best matched by position or name
+ * rather than by class.
*
- * @param oldSerializerSnapshot snapshot of the serializer that wrote the value.
- * @param value the value, already deserialized with the prior serializer.
+ * @param oldSerializerSnapshot snapshot of the serializer that wrote the value. A caller that
+ * holds the snapshot persisted with the state should pass that one in preference to a
+ * snapshot re-derived from a serializer restored from it, because that round trip does not
+ * always reproduce the schema that was written.
+ * @param value the value, already deserialized with the prior serializer. It may be {@code
+ * null} wherever the prior serializer can produce {@code null}. An implementation that
+ * decomposes a composite value may likewise pass {@code null} to a nested snapshot for an
+ * absent part, even where that part's serializer would reject {@code null} at top level.
* @return the value adapted to the schema of the current serializer.
*/
default T migrate(TypeSerializerSnapshot<T> oldSerializerSnapshot, T value) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java
index 2f50ba6c668..87cbfd23be4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java
@@ -26,6 +26,8 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.util.function.SupplierWithException;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.List;
import java.util.Map;
@@ -47,6 +49,13 @@ public class TtlAwareSerializer<T, S extends TypeSerializer<T>> extends TypeSeri
private final S typeSerializer;
+ /**
+ * Snapshot of {@link #bareValueSerializer()}, computed on first use. {@link
+ * #migrateValueFromPriorSerializer} runs once per migrated state value while the serializer
+ * stays the same, and taking a snapshot allocates one object per nested serializer.
+ */
+ private transient TypeSerializerSnapshot<?> bareValueSerializerSnapshot;
+
public TtlAwareSerializer(S typeSerializer) {
checkArgument(
!(typeSerializer instanceof TtlAwareSerializer),
@@ -128,31 +137,129 @@ public class TtlAwareSerializer<T, S extends TypeSerializer<T>> extends TypeSeri
return Objects.hash(isTtlEnabled, typeSerializer);
}
- @SuppressWarnings("unchecked")
+ /**
+ * Reads one state value written by {@code priorTtlAwareSerializer}, adapts it to this
+ * serializer's TTL setting and value schema, and writes it to {@code target}.
+ *
+ * <p>The value is unwrapped to its bare form, passed through {@link
+ * TypeSerializerSnapshot#migrate}, and re-wrapped. The hook returns the value unchanged unless
+ * the value serializer overrides it, so a value whose schema did not change is written back
+ * byte for byte.
+ *
+ * @param priorSerializerSnapshot the snapshot persisted with the state for {@code
+ * priorTtlAwareSerializer}, or {@code null} for a state that carries none.
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
public void migrateValueFromPriorSerializer(
TtlAwareSerializer<T, ?> priorTtlAwareSerializer,
+ @Nullable TypeSerializerSnapshot<T> priorSerializerSnapshot,
SupplierWithException<T, IOException> inputSupplier,
DataOutputView target,
TtlTimeProvider ttlTimeProvider)
throws IOException {
+ T priorValue = inputSupplier.get();
+ Object bareValue =
+ priorTtlAwareSerializer.wrapsTtlValue()
+ ? ((TtlValue<?>) priorValue).getUserValue()
+ : priorValue;
+
+ TypeSerializerSnapshot newSnapshot = bareValueSerializerSnapshot();
+ Object migratedValue =
+ newSnapshot.migrate(
+ priorBareValueSerializerSnapshot(
+ priorTtlAwareSerializer, priorSerializerSnapshot),
+ bareValue);
+
T outputRecord;
- if (this.isTtlEnabled()) {
- outputRecord =
- priorTtlAwareSerializer.isTtlEnabled
- ? inputSupplier.get()
- : (T)
- new TtlValue<>(
- inputSupplier.get(),
- ttlTimeProvider.currentTimestamp());
+ if (this.wrapsTtlValue()) {
+ // Carrying the prior timestamp over keeps the value's expiry where it was; migration
+ // is not a state access.
+ long lastAccessTimestamp =
+ priorTtlAwareSerializer.wrapsTtlValue()
+ ? ((TtlValue<?>) priorValue).getLastAccessTimestamp()
+ : ttlTimeProvider.currentTimestamp();
+ outputRecord = (T) new TtlValue<>(migratedValue, lastAccessTimestamp);
} else {
- outputRecord =
- priorTtlAwareSerializer.isTtlEnabled
- ? ((TtlValue<T>) inputSupplier.get()).getUserValue()
- : inputSupplier.get();
+ outputRecord = (T) migratedValue;
}
this.serialize(outputRecord, target);
}
+ /**
+ * The snapshot describing the schema the prior bare value was written with.
+ *
+ * <p>The snapshot persisted with the state is preferred over one re-derived from the prior
+ * serializer, because the prior serializer is itself restored from that snapshot and the round
+ * trip back to a snapshot is not always lossless: a POJO field that no longer exists on the
+ * class returns under a generated placeholder name, which would present a schema that was never
+ * written. Only the absence of a persisted snapshot falls back to the re-derived one: a
+ * persisted snapshot that does not match the prior serializer is an error, not a second reason
+ * to fall back, because re-deriving there would silently reintroduce that lossy round trip.
+ */
+ private static TypeSerializerSnapshot<?> priorBareValueSerializerSnapshot(
+ TtlAwareSerializer<?, ?> priorSerializer,
+ @Nullable TypeSerializerSnapshot<?> priorSerializerSnapshot) {
+ if (priorSerializerSnapshot == null) {
+ return priorSerializer.bareValueSerializerSnapshot();
+ }
+ // TtlAwareSerializerSnapshot is the snapshot counterpart of this class, so the persisted
+ // snapshot carries that layer wherever the serializer carries the wrapper: for a list or
+ // map state it is the element or value snapshot, for a value state the whole snapshot.
+ TypeSerializerSnapshot<?> priorSnapshot =
+ priorSerializerSnapshot instanceof TtlAwareSerializerSnapshot
+ ? ((TtlAwareSerializerSnapshot<?>) priorSerializerSnapshot)
+ .getOriginalTypeSerializerSnapshot()
+ : priorSerializerSnapshot;
+
+ // Thrown rather than checked through Preconditions: this runs once per migrated state
+ // value, so the message must not be built while the check is passing.
+ boolean isTtlSnapshot = priorSnapshot instanceof TtlStateFactory.TtlSerializerSnapshot;
+ if (!priorSerializer.wrapsTtlValue()) {
+ if (isTtlSnapshot) {
+ throw new IllegalArgumentException(
+ "The prior serializer does not wrap values in TtlValue, but its persisted snapshot is a TtlSerializerSnapshot.");
+ }
+ return priorSnapshot;
+ }
+ if (!isTtlSnapshot) {
+ throw new IllegalArgumentException(
+ "The prior serializer wraps values in TtlValue, so its persisted snapshot should be a TtlSerializerSnapshot, but was "
+ + priorSnapshot.getClass().getName()
+ + ".");
+ }
+ // The persisted snapshot describes the TtlValue envelope, so descend to the user value
+ // the same way bareValueSerializer() descends the serializer.
+ return ((TtlStateFactory.TtlSerializerSnapshot<?>) priorSnapshot)
+ .getValueSerializerSnapshot();
+ }
+
+ private TypeSerializerSnapshot<?> bareValueSerializerSnapshot() {
+ if (bareValueSerializerSnapshot == null) {
+ bareValueSerializerSnapshot = bareValueSerializer().snapshotConfiguration();
+ }
+ return bareValueSerializerSnapshot;
+ }
+
+ /**
+ * The serializer of the bare (non-TTL) value: the user value serializer of a {@link
+ * TtlStateFactory.TtlSerializer}, otherwise the wrapped serializer itself.
+ */
+ private TypeSerializer<?> bareValueSerializer() {
+ return wrapsTtlValue()
+ ? ((TtlStateFactory.TtlSerializer<?>) typeSerializer).getValueSerializer()
+ : typeSerializer;
+ }
+
+ /**
+ * Whether the values this serializer reads and writes are {@link TtlValue} envelopes. Narrower
+ * than {@link #isTtlEnabled()}, which is also true for a list or map serializer whose element
+ * or value serializer is a {@link TtlStateFactory.TtlSerializer}: such a serializer wraps the
+ * collection, not a single {@code TtlValue}.
+ */
+ private boolean wrapsTtlValue() {
+ return typeSerializer instanceof TtlStateFactory.TtlSerializer;
+ }
+
@Override
public void copy(DataInputView source, DataOutputView target) throws IOException {
typeSerializer.copy(source, target);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java
index 65908573fac..96b4479c915 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java
@@ -21,13 +21,18 @@ package org.apache.flink.runtime.state;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.runtime.state.ttl.TtlAwareSerializerSnapshot;
import org.apache.flink.runtime.testutils.statemigration.TestType;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -311,6 +316,42 @@ class StateSerializerProviderTest {
.isInstanceOf(IllegalStateException.class);
}
+ // --------------------------------------------------------------------------------
+ // Tests for the ttl-aware wrapping of the previous serializer snapshot
+ // --------------------------------------------------------------------------------
+
+ /**
+ * Registering a new serializer replaces the nested snapshot of the previous snapshot in place,
+ * so the snapshot a caller still holds no longer reports what the checkpoint wrote: its nested
+ * snapshot becomes a {@link TtlAwareSerializerSnapshot} around the original. Anything that
+ * descends a restored composite snapshot has to expect that layer.
+ */
+ @Test
+ void testRegisterNewSerializerWrapsNestedSnapshotOfPreviousSnapshotInPlace() {
+ ListSerializerSnapshot<String> previousSnapshot =
+ (ListSerializerSnapshot<String>)
+ new ListSerializer<>(StringSerializer.INSTANCE).snapshotConfiguration();
+ TypeSerializerSnapshot<String> elementSnapshotAsWritten =
+ previousSnapshot.getElementSerializerSnapshot();
+
+ StateSerializerProvider<List<String>> testProvider =
+ StateSerializerProvider.fromPreviousSerializerSnapshot(previousSnapshot);
+ testProvider.registerNewSerializerForRestoredState(
+ new ListSerializer<>(StringSerializer.INSTANCE));
+
+ // The same snapshot instance now reports a different element snapshot than it did above.
+ TypeSerializerSnapshot<String> elementSnapshotAfterRestore =
+ previousSnapshot.getElementSerializerSnapshot();
+ assertThat(elementSnapshotAfterRestore).isNotSameAs(elementSnapshotAsWritten);
+ assertThat(elementSnapshotAfterRestore).isInstanceOf(TtlAwareSerializerSnapshot.class);
+ // The original is carried inside the wrapper, not re-derived: a re-derived snapshot would
+ // be an equal instance of the same class but a different object.
+ assertThat(
+ ((TtlAwareSerializerSnapshot<String>) elementSnapshotAfterRestore)
+ .getOriginalTypeSerializerSnapshot())
+ .isSameAs(elementSnapshotAsWritten);
+ }
+
// --------------------------------------------------------------------------------
// Utilities
// --------------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java
index f91e71d360c..8e1d376cd09 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java
@@ -18,19 +18,36 @@
package org.apache.flink.runtime.state.ttl;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.ListSerializer;
import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.common.typeutils.base.MapSerializer;
import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.DataOutputView;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
+
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
class TtlAwareSerializerTest {
+ private static final String VALUE = "value";
+ private static final long PRIOR_TIMESTAMP = 1_000L;
+ private static final long CURRENT_TIMESTAMP = 9_999L;
+ private static final TtlTimeProvider FIXED_TIME_PROVIDER = () -> CURRENT_TIMESTAMP;
+
@Test
void testSerializerTtlEnabled() {
IntSerializer intSerializer = IntSerializer.INSTANCE;
@@ -147,4 +164,385 @@ class TtlAwareSerializerTest {
.getValueSerializerSnapshot()))
.isInstanceOf(TtlAwareSerializerSnapshot.class);
}
+
+ @Test
+ void testMigrateValueNoTtlToNoTtl() throws IOException {
+ TtlAwareSerializer<?, ?> prior = stringSerializer(false);
+ TtlAwareSerializer<?, ?> current = stringSerializer(false);
+
+ assertThat(migrate(current, prior, VALUE)).isEqualTo(serialize(current, VALUE));
+ }
+
+ @Test
+ void testMigrateValueNoTtlToTtlStampsCurrentTime() throws IOException {
+ TtlAwareSerializer<?, ?> prior = stringSerializer(false);
+ TtlAwareSerializer<?, ?> current = stringSerializer(true);
+
+ assertThat(migrate(current, prior, VALUE))
+ .isEqualTo(serialize(current, new TtlValue<>(VALUE, CURRENT_TIMESTAMP)));
+ }
+
+ @Test
+ void testMigrateValueTtlToNoTtlUnwraps() throws IOException {
+ TtlAwareSerializer<?, ?> prior = stringSerializer(true);
+ TtlAwareSerializer<?, ?> current = stringSerializer(false);
+
+ assertThat(migrate(current, prior, new TtlValue<>(VALUE, PRIOR_TIMESTAMP)))
+ .isEqualTo(serialize(current, VALUE));
+ }
+
+ @Test
+ void testMigrateValueTtlToTtlKeepsPriorTimestamp() throws IOException {
+ TtlAwareSerializer<?, ?> prior = stringSerializer(true);
+ TtlAwareSerializer<?, ?> current = stringSerializer(true);
+ TtlValue<String> priorValue = new TtlValue<>(VALUE, PRIOR_TIMESTAMP);
+
+ assertThat(migrate(current, prior, priorValue)).isEqualTo(serialize(current, priorValue));
+ }
+
+ @Test
+ void testMigrateValueInvokesHookOnNewSnapshotWithOldSnapshotAsArgument() throws IOException {
+ TtlAwareSerializer<?, ?> prior =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old"));
+ TtlAwareSerializer<?, ?> current =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new"));
+
+ byte[] migrated = migrate(current, prior, VALUE);
+
+ assertThat(current.deserialize(new DataInputDeserializer(migrated)))
+ .isEqualTo(VALUE + "|from=old|to=new");
+ }
+
+ @Test
+ void testMigrateValueInvokesHookOnUnwrappedValueWithInnerSnapshots() throws IOException {
+ TtlAwareSerializer<?, ?> prior = taggedTtlSerializer("old");
+ TtlAwareSerializer<?, ?> current = taggedTtlSerializer("new");
+
+ byte[] migrated = migrate(current, prior, new TtlValue<>(VALUE, PRIOR_TIMESTAMP));
+
+ TtlValue<?> result = (TtlValue<?>) current.deserialize(new DataInputDeserializer(migrated));
+ assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=old|to=new");
+ assertThat(result.getLastAccessTimestamp()).isEqualTo(PRIOR_TIMESTAMP);
+ }
+
+ @Test
+ void testMigrateValueUsesPersistedPriorSnapshotNotOneRederivedFromPriorSerializer()
+ throws IOException {
+ TtlAwareSerializer<?, ?> prior =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("rederived"));
+ TtlAwareSerializer<?, ?> current =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new"));
+
+ byte[] migrated = migrate(current, prior, new TaggedSnapshot("persisted"), VALUE);
+
+ assertThat(current.deserialize(new DataInputDeserializer(migrated)))
+ .isEqualTo(VALUE + "|from=persisted|to=new");
+ }
+
+ @Test
+ void testMigrateValueDescendsPersistedTtlSnapshotToItsValueSnapshot() throws IOException {
+ TtlAwareSerializer<?, ?> prior = taggedTtlSerializer("rederived");
+ TtlAwareSerializer<?, ?> current = taggedTtlSerializer("new");
+ // Tagged differently from the prior serializer, so both falling back to a re-derived
+ // snapshot and skipping the descent into the TtlValue envelope change the result.
+ TypeSerializerSnapshot<?> persisted =
+ taggedTtlSerializer("persisted")
+ .getOriginalTypeSerializer()
+ .snapshotConfiguration();
+
+ byte[] migrated =
+ migrate(current, prior, persisted, new TtlValue<>(VALUE, PRIOR_TIMESTAMP));
+
+ TtlValue<?> result = (TtlValue<?>) current.deserialize(new DataInputDeserializer(migrated));
+ assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=persisted|to=new");
+ }
+
+ /**
+ * A list or map state persists its element or value snapshot wrapped in a {@link
+ * TtlAwareSerializerSnapshot}, which the descent has to see through to reach the user value
+ * snapshot. Value state persists the unwrapped shape covered by the test above.
+ */
+ @Test
+ void testMigrateValueDescendsPersistedTtlAwareElementSnapshot() throws IOException {
+ ListSerializerSnapshot<?> persistedListSnapshot =
+ (ListSerializerSnapshot<?>)
+ TtlAwareSerializer.wrapTtlAwareSerializer(
+ new ListSerializer<>(rawTaggedTtlSerializer("persisted")))
+ .snapshotConfiguration();
+ TypeSerializerSnapshot<?> persistedElementSnapshot =
+ persistedListSnapshot.getElementSerializerSnapshot();
+ assertThat(persistedElementSnapshot).isInstanceOf(TtlAwareSerializerSnapshot.class);
+
+ TtlAwareSerializer<?, ?> prior = taggedTtlSerializer("rederived");
+ TtlAwareSerializer<?, ?> current = taggedTtlSerializer("new");
+
+ byte[] migrated =
+ migrate(
+ current,
+ prior,
+ persistedElementSnapshot,
+ new TtlValue<>(VALUE, PRIOR_TIMESTAMP));
+
+ TtlValue<?> result = (TtlValue<?>) current.deserialize(new DataInputDeserializer(migrated));
+ assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=persisted|to=new");
+ }
+
+ @Test
+ void testMigrateValueRejectsNonTtlSnapshotForTtlPriorSerializer() {
+ TtlAwareSerializer<?, ?> prior = taggedTtlSerializer("old");
+ TtlAwareSerializer<?, ?> current = taggedTtlSerializer("new");
+
+ assertThatThrownBy(
+ () ->
+ migrate(
+ current,
+ prior,
+ new TaggedSnapshot("mismatched"),
+ new TtlValue<>(VALUE, PRIOR_TIMESTAMP)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be a TtlSerializerSnapshot");
+ }
+
+ @Test
+ void testMigrateValueRejectsNonTtlSnapshotBehindTtlAwareLayer() {
+ TtlAwareSerializer<?, ?> prior = taggedTtlSerializer("old");
+ TtlAwareSerializer<?, ?> current = taggedTtlSerializer("new");
+ // The TtlAware layer is seen through, so a mismatch inside it is still caught.
+ TypeSerializerSnapshot<?> wrappedMismatch =
+ new TtlAwareSerializerSnapshot<>(new TaggedSnapshot("mismatched"));
+
+ assertThatThrownBy(
+ () ->
+ migrate(
+ current,
+ prior,
+ wrappedMismatch,
+ new TtlValue<>(VALUE, PRIOR_TIMESTAMP)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be a TtlSerializerSnapshot");
+ }
+
+ @Test
+ void testMigrateValueRejectsTtlSnapshotForNonTtlPriorSerializer() {
+ TtlAwareSerializer<?, ?> prior =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old"));
+ TtlAwareSerializer<?, ?> current =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new"));
+ TypeSerializerSnapshot<?> ttlSnapshot =
+ rawTaggedTtlSerializer("mismatched").snapshotConfiguration();
+
+ assertThatThrownBy(() -> migrate(current, prior, ttlSnapshot, VALUE))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("does not wrap values in TtlValue");
+ }
+
+ @Test
+ void testMigrateValueFallsBackToPriorSerializerSnapshotWhenNonePersisted() throws IOException {
+ TtlAwareSerializer<?, ?> prior =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old"));
+ TtlAwareSerializer<?, ?> current =
+ TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new"));
+
+ byte[] migrated = migrate(current, prior, null, VALUE);
+
+ assertThat(current.deserialize(new DataInputDeserializer(migrated)))
+ .isEqualTo(VALUE + "|from=old|to=new");
+ }
+
+ @Test
+ void testMigrateValueTtlToTtlWithNullUserValue() throws IOException {
+ TtlAwareSerializer<?, ?> prior = nullTolerantTtlSerializer();
+ TtlAwareSerializer<?, ?> current = nullTolerantTtlSerializer();
+
+ byte[] migrated = migrate(current, prior, new TtlValue<>(null, PRIOR_TIMESTAMP));
+
+ TtlValue<?> result = (TtlValue<?>) current.deserialize(new DataInputDeserializer(migrated));
+ assertThat(result.getUserValue()).isNull();
+ assertThat(result.getLastAccessTimestamp()).isEqualTo(PRIOR_TIMESTAMP);
+ }
+
+ /** Migrates with the snapshot the state backends would have persisted for {@code prior}. */
+ private static byte[] migrate(
+ TtlAwareSerializer<?, ?> current, TtlAwareSerializer<?, ?> prior, Object priorValue)
+ throws IOException {
+ return migrate(
+ current,
+ prior,
+ prior.getOriginalTypeSerializer().snapshotConfiguration(),
+ priorValue);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static byte[] migrate(
+ TtlAwareSerializer<?, ?> current,
+ TtlAwareSerializer<?, ?> prior,
+ TypeSerializerSnapshot<?> priorSnapshot,
+ Object priorValue)
+ throws IOException {
+ DataOutputSerializer output = new DataOutputSerializer(64);
+ ((TtlAwareSerializer) current)
+ .migrateValueFromPriorSerializer(
+ (TtlAwareSerializer) prior,
+ priorSnapshot,
+ () -> priorValue,
+ output,
+ FIXED_TIME_PROVIDER);
+ return output.getCopyOfBuffer();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static byte[] serialize(TtlAwareSerializer<?, ?> serializer, Object value)
+ throws IOException {
+ DataOutputSerializer output = new DataOutputSerializer(64);
+ ((TtlAwareSerializer) serializer).serialize(value, output);
+ return output.getCopyOfBuffer();
+ }
+
+ private static TtlAwareSerializer<?, ?> stringSerializer(boolean ttlEnabled) {
+ return ttlEnabled
+ ? TtlAwareSerializer.wrapTtlAwareSerializer(
+ new TtlStateFactory.TtlSerializer<>(
+ LongSerializer.INSTANCE, StringSerializer.INSTANCE))
+ : TtlAwareSerializer.wrapTtlAwareSerializer(StringSerializer.INSTANCE);
+ }
+
+ private static TtlAwareSerializer<?, ?> taggedTtlSerializer(String tag) {
+ return TtlAwareSerializer.wrapTtlAwareSerializer(rawTaggedTtlSerializer(tag));
+ }
+
+ private static TtlStateFactory.TtlSerializer<String> rawTaggedTtlSerializer(String tag) {
+ return new TtlStateFactory.TtlSerializer<>(
+ LongSerializer.INSTANCE, new TaggedStringSerializer(tag));
+ }
+
+ private static TtlAwareSerializer<?, ?> nullTolerantTtlSerializer() {
+ return TtlAwareSerializer.wrapTtlAwareSerializer(
+ new TtlStateFactory.TtlSerializer<>(
+ LongSerializer.INSTANCE,
+ NullableSerializer.wrapIfNullIsNotSupported(
+ StringSerializer.INSTANCE, false)));
+ }
+
+ /** A string serializer whose snapshot records the schema it belongs to. */
+ private static final class TaggedStringSerializer extends TypeSerializer<String> {
+
+ private final String tag;
+
+ private TaggedStringSerializer(String tag) {
+ this.tag = tag;
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public TypeSerializer<String> duplicate() {
+ return this;
+ }
+
+ @Override
+ public String createInstance() {
+ return "";
+ }
+
+ @Override
+ public String copy(String from) {
+ return from;
+ }
+
+ @Override
+ public String copy(String from, String reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(String record, DataOutputView target) throws IOException {
+ StringSerializer.INSTANCE.serialize(record, target);
+ }
+
+ @Override
+ public String deserialize(DataInputView source) throws IOException {
+ return StringSerializer.INSTANCE.deserialize(source);
+ }
+
+ @Override
+ public String deserialize(String reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ StringSerializer.INSTANCE.copy(source, target);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof TaggedStringSerializer
+ && tag.equals(((TaggedStringSerializer) obj).tag);
+ }
+
+ @Override
+ public int hashCode() {
+ return tag.hashCode();
+ }
+
+ @Override
+ public TypeSerializerSnapshot<String> snapshotConfiguration() {
+ return new TaggedSnapshot(tag);
+ }
+ }
+
+ /**
+ * Appends both the argument snapshot's tag and its own tag to the migrated value, so a
+ * migration that swaps receiver and argument produces a different result.
+ */
+ public static final class TaggedSnapshot implements TypeSerializerSnapshot<String> {
+
+ private String tag;
+
+ public TaggedSnapshot() {}
+
+ private TaggedSnapshot(String tag) {
+ this.tag = tag;
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ out.writeUTF(tag);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+ throws IOException {
+ tag = in.readUTF();
+ }
+
+ @Override
+ public TypeSerializer<String> restoreSerializer() {
+ return new TaggedStringSerializer(tag);
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<String> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<String> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+
+ @Override
+ public String migrate(TypeSerializerSnapshot<String> oldSerializerSnapshot, String value) {
+ return value + "|from=" + ((TaggedSnapshot) oldSerializerSnapshot).tag + "|to=" + tag;
+ }
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java
index d8b4234252c..ffe2f41c03a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java
@@ -161,6 +161,7 @@ public class TtlAwareSerializerUpgradeTest
DataOutputSerializer migratedOut = new DataOutputSerializer(INITIAL_OUTPUT_BUFFER_SIZE);
writer.migrateValueFromPriorSerializer(
reader,
+ reader.getOriginalTypeSerializer().snapshotConfiguration(),
() -> reader.deserialize(originalDataInput),
migratedOut,
TtlTimeProvider.DEFAULT);
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java
index b60347ff208..758f7123793 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java
@@ -19,6 +19,7 @@ package org.apache.flink.state.rocksdb;
import org.apache.flink.api.common.state.State;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
@@ -36,6 +37,8 @@ import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.RocksDBException;
import org.rocksdb.WriteOptions;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -191,6 +194,7 @@ public abstract class AbstractRocksDBState<K, N, V> implements InternalKvState<K
DataInputDeserializer serializedOldValueInput,
DataOutputSerializer serializedMigratedValueOutput,
TypeSerializer<V> priorSerializer,
+ @Nullable TypeSerializerSnapshot<V> priorSerializerSnapshot,
TypeSerializer<V> newSerializer,
TtlTimeProvider ttlTimeProvider)
throws StateMigrationException {
@@ -203,6 +207,7 @@ public abstract class AbstractRocksDBState<K, N, V> implements InternalKvState<K
try {
ttlAwareNewSerializer.migrateValueFromPriorSerializer(
ttlAwarePriorSerializer,
+ priorSerializerSnapshot,
() -> ttlAwarePriorSerializer.deserialize(serializedOldValueInput),
serializedMigratedValueOutput,
ttlTimeProvider);
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java
index 73142571373..c6512f4a356 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java
@@ -916,9 +916,12 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N, SV>> stateMetaInfo)
throws Exception {
+ // The snapshot the state was written with. Preferred over one re-derived from the previous
+ // serializer, which is itself restored from this snapshot and does not always round trip.
+ TypeSerializerSnapshot<SV> previousSerializerSnapshot =
+ stateMetaInfo.f1.getPreviousStateSerializerSnapshot();
+
if (stateDesc.getType() == StateDescriptor.Type.MAP) {
- TypeSerializerSnapshot<SV> previousSerializerSnapshot =
- stateMetaInfo.f1.getPreviousStateSerializerSnapshot();
checkState(
previousSerializerSnapshot != null,
"the previous serializer snapshot should exist.");
@@ -1003,6 +1006,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
serializedValueInput,
migratedSerializedValueOutput,
previousTtlAwareSerializer,
+ previousSerializerSnapshot,
currentTtlAwareSerializer,
this.ttlTimeProvider);
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java
index 8c6088df23d..66b273a5ecb 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java
@@ -22,7 +22,9 @@ import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.State;
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
@@ -200,6 +202,7 @@ class RocksDBListState<K, N, V> extends AbstractRocksDBState<K, N, List<V>>
DataInputDeserializer serializedOldValueInput,
DataOutputSerializer serializedMigratedValueOutput,
TypeSerializer<List<V>> priorSerializer,
+ @Nullable TypeSerializerSnapshot<List<V>> priorSerializerSnapshot,
TypeSerializer<List<V>> newSerializer,
TtlTimeProvider ttlTimeProvider)
throws StateMigrationException {
@@ -214,11 +217,29 @@ class RocksDBListState<K, N, V> extends AbstractRocksDBState<K, N, List<V>>
TtlAwareSerializer<V, ?> newTtlAwareElementSerializer =
((TtlAwareSerializer.TtlAwareListSerializer<V>) newSerializer)
.getElementSerializer();
+ // Descend the persisted snapshot the same way as the serializer, so element migration
+ // sees the schema the elements were written with. A state that carries no persisted
+ // snapshot leaves this null, and the element migration re-derives one instead.
+ TypeSerializerSnapshot<V> priorElementSerializerSnapshot = null;
+ if (priorSerializerSnapshot != null) {
+ // Thrown rather than checked through Preconditions: this method runs once per state
+ // entry, so the message must not be built while the check is passing.
+ if (!(priorSerializerSnapshot instanceof ListSerializerSnapshot)) {
+ throw new IllegalArgumentException(
+ "The previous serializer snapshot of a list state should be a ListSerializerSnapshot, but was "
+ + priorSerializerSnapshot.getClass().getName()
+ + ".");
+ }
+ priorElementSerializerSnapshot =
+ ((ListSerializerSnapshot<V>) priorSerializerSnapshot)
+ .getElementSerializerSnapshot();
+ }
try {
while (serializedOldValueInput.available() > 0) {
newTtlAwareElementSerializer.migrateValueFromPriorSerializer(
priorTtlAwareElementSerializer,
+ priorElementSerializerSnapshot,
() ->
ListDelimitedSerializer.deserializeNextElement(
serializedOldValueInput, priorTtlAwareElementSerializer),
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java
index 8f33bfe6aad..6dfd00aaca1 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java
@@ -23,7 +23,9 @@ import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.State;
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.MapSerializer;
+import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
@@ -231,6 +233,7 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U
DataInputDeserializer serializedOldValueInput,
DataOutputSerializer serializedMigratedValueOutput,
TypeSerializer<Map<UK, UV>> priorSerializer,
+ @Nullable TypeSerializerSnapshot<Map<UK, UV>> priorSerializerSnapshot,
TypeSerializer<Map<UK, UV>> newSerializer,
TtlTimeProvider ttlTimeProvider)
throws StateMigrationException {
@@ -244,6 +247,23 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U
TtlAwareSerializer<UV, ?> newTtlAwareMapValueSerializer =
((TtlAwareSerializer.TtlAwareMapSerializer<UK, UV>) newSerializer)
.getValueSerializer();
+ // Descend the persisted snapshot the same way as the serializer, so value migration sees
+ // the schema the map values were written with. A state that carries no persisted
+ // snapshot leaves this null, and the value migration re-derives one instead.
+ TypeSerializerSnapshot<UV> priorMapValueSerializerSnapshot = null;
+ if (priorSerializerSnapshot != null) {
+ // Thrown rather than checked through Preconditions: this method runs once per state
+ // entry, so the message must not be built while the check is passing.
+ if (!(priorSerializerSnapshot instanceof MapSerializerSnapshot)) {
+ throw new IllegalArgumentException(
+ "The previous serializer snapshot of a map state should be a MapSerializerSnapshot, but was "
+ + priorSerializerSnapshot.getClass().getName()
+ + ".");
+ }
+ priorMapValueSerializerSnapshot =
+ ((MapSerializerSnapshot<UK, UV>) priorSerializerSnapshot)
+ .getValueSerializerSnapshot();
+ }
try {
boolean isNull = serializedOldValueInput.readBoolean();
@@ -254,6 +274,7 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U
} else {
newTtlAwareMapValueSerializer.migrateValueFromPriorSerializer(
priorTtlAwareMapValueSerializer,
+ priorMapValueSerializerSnapshot,
() -> priorTtlAwareMapValueSerializer.deserialize(serializedOldValueInput),
serializedMigratedValueOutput,
ttlTimeProvider);