在flink双流Time-windowed Joins的主要实现是在TimeBoundedStreamJoin中，这个类里面的变量非常的多，所以首先要清楚，这些重要变量或者概念的计算过程。简单的说整个join过程就是把左流的数据和右流的数据都通过state保存起来，左流有新的数据到，就会根据key去遍历右流state中的数据，符合关联条件就输出，关联不上的就保存在左流的state中等待右流数据的遍历，反之亦然。另外会对每个流计算过期时间，以及每个数据的清理时间。本文主要根据代码的实现过程对清理机制做一个分步的演算。

基本公式

建表语句

CREATE TABLE LeftTable (
  l_id STRING, 
  l_imsi STRING, 
  l_time TIMESTAMP(3), 
  WATERMARK FOR l_time AS l_time - INTERVAL '5' SECOND 
) WITH ( 
  'connector.type' = 'kafka',   
  'connector.version' = 'universal',  
  'connector.topic' = 'foo',  
  'connector.properties.zookeeper.connect' = 'hostA:2181',
  'connector.properties.bootstrap.servers' = 'hostA:6667',  
  'connector.properties.group.id' = 'tg1',   
  'connector.startup-mode' = 'latest-offset', 
  'format.type' = 'csv',   
  'format.field-delimiter' = ','
  )

CREATE TABLE RightTable (
  r_id STRING, 
  r_location STRING, 
  r_time TIMESTAMP(3), 
  WATERMARK FOR r_time AS r_time - INTERVAL '2' SECOND 
) WITH (  
  'connector.type' = 'kafka',   
  'connector.version' = 'universal',  
  'connector.topic' = 'bar',  
  'connector.properties.zookeeper.connect' = 'hostA:2181',
  'connector.properties.bootstrap.servers' = 'hostA:6667',
  'connector.properties.group.id' = 'tg1',   
  'connector.startup-mode' = 'latest-offset', 
  'format.type' = 'csv',   
  'format.field-delimiter' = ',' 
)

执行SQL

SELECT l_id, l_imsi, r_location 
FROM LeftTable 
LEFT JOIN RightTable 
on l_id = r_id 
and r_time >= l_time - INTERVAL '4' SECOND AND r_time <= l_time + INTERVAL '6' SECOND

即 upperBound = 6和lowerNound = -4

那么

leftRelativeSize = -leftLowerBound = -(-upperBound) = upperBound = 6

rightRelativeSize = leftUpperBound = -lowerBound = 4

leftExpirationTime = wm - upperBound - 0.001 
                   = wm - 6 - 0.001 
                   = wm - 6.001

rightExpirationTime = wm + lowerBound - 0.001 
                    = wm - 4 - 0.001 
                    = wm - 4.001
  
leftRowCleanUpTime = rowTime + leftRelativeSize + minCleanUpInterval + allowedLateness + 0.001 
                   = rowTime + 6 + (6+4)/2 + 0.001 
                   = rowTime + 11.001
                   
rightRowCleanUpTime = rowTime + rightRelativeSize + minCleanUpInterval + allowedLateness + 0.001 
                    = rowTime + 4 + (6+4)/2 + 0.001 
                    = rowTime + 9.001

rightOperatorTime = leftOperatorTime = wm = min(leftWatermark, rightWatermark)
  
allowedLateness = 0 //忽略不计

根据上面的公式进行模拟计算

总结

每次计算ExpirationTime的时候用的是上一次的wm；
6-1，6-2，6-3为三种独立可能触发删除过期数据的场景；
通过上面的测试可以发现，数据即便过期了，但是没有到清理时间，如果这时候有符合关联条件的数据还是可以关联上的，例如6-2场景。

以上数据基于flink 1.10.0版本blink planner进行测试。