[FLINK-10674] Fix handling of retractions after clean up

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.6.1
• Fix Version/s: 1.5.6, 1.6.3, 1.7.0
• Component/s: Table SQL / API
• Labels: pull-request-available

Environment

Flink 1.6.0

• Assignee: Timo Walther
• Reporter: ambition
• Created: 25/Oct/18 06:52
• Updated: 27/Jul/20 12:58
• Resolved: 22/Nov/18 15:22

Description

Our online Flink Job run about a week，job contain sql ：

select  `time`,  
        lower(trim(os_type)) as os_type, 
        count(distinct feed_id) as feed_total_view  
from  my_table 
group by `time`, lower(trim(os_type))

 

  then occur NPE: 

 

java.lang.NullPointerException

at scala.Predef$.Long2long(Predef.scala:363)

at org.apache.flink.table.functions.aggfunctions.DistinctAccumulator.remove(DistinctAccumulator.scala:109)

at NonWindowedAggregationHelper$894.retract(Unknown Source)

at org.apache.flink.table.runtime.aggregate.GroupAggProcessFunction.processElement(GroupAggProcessFunction.scala:124)

at org.apache.flink.table.runtime.aggregate.GroupAggProcessFunction.processElement(GroupAggProcessFunction.scala:39)

at org.apache.flink.streaming.api.operators.LegacyKeyedProcessOperator.processElement(LegacyKeyedProcessOperator.java:88)

at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202)

at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:105)

at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)

at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)

at java.lang.Thread.run(Thread.java:745)

 

 

View DistinctAccumulator.remove

 

this NPE should currentCnt = null lead to, so we simple handle like :

def remove(params: Row): Boolean = {
  if(!distinctValueMap.contains(params)){
    true
  }else{
    val currentCnt = distinctValueMap.get(params)
    // 
    if (currentCnt == null || currentCnt == 1) {
      distinctValueMap.remove(params)
      true
    } else {
      var value = currentCnt - 1L
      if(value < 0){
        value = 1
      }
      distinctValueMap.put(params, value)
      false
    }
  }
}

 

Update:

Because state clean up happens in processing time, it might be
the case that retractions are arriving after the state has
been cleaned up. Before these changes, a new accumulator was
created and invalid retraction messages were emitted. This
change drops retraction messages for which no accumulator
exists. 

Issue Links

• links to: GitHub Pull Request #7067
• links to: GitHub Pull Request #7076
• links to: GitHub Pull Request #7147