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.



---

Reply via email to