Github user sihuazhou commented on a diff in the pull request: https://github.com/apache/flink/pull/6186#discussion_r197111980 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlValue.java --- @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.state.ttl; + +import org.apache.flink.util.Preconditions; + +import java.io.Serializable; + +/** + * This class wraps user value of state with TTL. + * + * @param <T> Type of the user value of state with TTL + */ +class TtlValue<T> implements Serializable { + private final T userValue; + private final long expirationTimestamp; --- End diff -- I'm really not sure whether we should leave it for now. If we leave it for now, then it will be a headache problem on practical production. As a very common situation there is a job, which reading data from kafka, and the user set the `TTL = 2hours` because he thinks that the data's latency is absolute less than 2 hours, this way they can use the TTL to safely control the whole state size, and got a exactly result. But, if he found that the job need to scale up, then he need to trigger a savepoint and rescale the job from it. but what if there's some problems that stop he recovering the job from the savepoint in a very short time, let's say he will took 30min to recover the job, then the result become inaccuracy. Even the user never need to trigger a savepoint for any reason, what if the job means some problem(maybe some problem with some machine) and loop in "failed-restart-failed-..", after 2 hours we fixed the problem and the job automatically resume, but the state has all been expired. I think this is a disaster for the user. Yes, when using the `EventTime` people this problem won't help, but the `ProccessTime` is a very common use case(In our production, most of the job's TimeCharacter is `ProccessTime`). I know Flink's TimeService also didn't do the time align works on recovery, but state's TTL is a bit different with Timer. When registering a timer, what users offer to the API is a absolute time, but when setting the TTL, what users offer is just a relative time, it's us that convert the relative time to a absolute time to implement the TTL.
---