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 <[email protected]>
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);
