Copilot commented on code in PR #28881:
URL: https://github.com/apache/flink/pull/28881#discussion_r3888287049
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java:
##########
@@ -214,11 +217,29 @@ public void migrateSerializedValue(
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()
+ + ".");
+ }
Review Comment:
This type-guard throws IllegalArgumentException outside the try/catch, so an
unexpected priorSerializerSnapshot type will bypass the method’s
StateMigrationException wrapping and escape as an unchecked exception. Prefer
throwing StateMigrationException (or moving the guard into the try) so restore
failures remain consistently categorized as state migration failures.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java:
##########
@@ -128,31 +137,129 @@ public int hashCode() {
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)
+ .getOrinalTypeSerializerSnapshot()
+ : priorSerializerSnapshot;
Review Comment:
The accessor name getOrinalTypeSerializerSnapshot() is misspelled ("Orinal")
and the typo is now being propagated into new production code paths. Consider
adding a correctly spelled getOriginalTypeSerializerSnapshot() (keeping the old
method as a deprecated alias for compatibility) and switching call sites to the
corrected name to avoid locking in the typo.
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java:
##########
@@ -240,6 +243,23 @@ public void migrateSerializedValue(
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()
+ + ".");
+ }
Review Comment:
This type-guard throws IllegalArgumentException before entering the
try/catch, so a snapshot-type mismatch will not be wrapped into the
StateMigrationException that callers of migrateSerializedValue likely expect.
Throw StateMigrationException directly (or move the guard into the try) to keep
restore failures consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
