[FLINK-10567][state] Fix TtlStateSerializer doe not propagate all fie…

…ld serializer during duplicate() This closes apache#6860.
laubersder · Oct 18, 2018 · 4be1afa · 4be1afa
1 parent d5aa97f
commit 4be1afa
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 1 deletion.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlStateFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlStateFactory.java
@@ -224,7 +224,7 @@ protected CompositeSerializer<TtlValue<T>> createSerializerInstance(
  TypeSerializer<?> ... originalSerializers) {
  Preconditions.checkNotNull(originalSerializers);
  Preconditions.checkArgument(originalSerializers.length == 2);
- return new TtlSerializer<>(precomputed, (TypeSerializer<T>) originalSerializers[1]);
+ return new TtlSerializer<>(precomputed, originalSerializers);
  }
  }
 }
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
@@ -33,8 +33,10 @@
 import org.apache.flink.api.common.state.ReducingState;
 import org.apache.flink.api.common.state.ReducingStateDescriptor;
 import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
 import org.apache.flink.api.common.state.ValueState;
 import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompatibilityResult;
 import org.apache.flink.api.common.typeutils.ParameterlessTypeSerializerConfig;
@@ -1341,6 +1343,23 @@ public void testValueState() throws Exception {
  backend.dispose();
  }
 
+ @Test
+ public void testValueStateWorkWithTtl() throws Exception {
+ AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE);
+ try {
+ ValueStateDescriptor<MutableLong> kvId = new ValueStateDescriptor<>("id", MutableLong.class);
+ kvId.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(1)).build());
+
+ ValueState<MutableLong> state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
+ backend.setCurrentKey(1);
+ state.update(new MutableLong());
+ state.value();
+ } finally {
+ backend.close();
+ backend.dispose();
+ }
+ }
+
  /**
  * Tests {@link ValueState#value()} and
  * {@link InternalKvState#getSerializedValue(byte[], TypeSerializer, TypeSerializer, TypeSerializer)}