Getting error: task start: cannot add child mismatched edges: mean5:stream -> join9:batch

Store
1

I have data being collected in testdata db with a delay of 5 minutes. So at the current moment, there is no data and the data would be available after 5 minutes for current moment.

Now I am trying to clean up data cause there is noise and I am trying to iron out spikes, for same following is the tick script:

dbrp "testdata"."autogen"
var metric = 'liters'

var historical = stream
  |from()
    .database('testdata')
    .retentionPolicy('autogen')
    .measurement('overhead')
    .where(lambda: TRUE)
  |window()
    .period(15m)
    .every(30s)
    .align()

var actual = stream
  |from()
    .database('testdata')
    .retentionPolicy('autogen')
    .measurement('overhead')
    .where(lambda: TRUE)
  |window()
    .period(15m)
    .every(30s)
    .align()

var avg = actual|mean(metric).as('value') 
    
historical
  |shift(1m)
  |join(actual, avg)
.as('historical', 'actual', 'avg')
.fill('null')
.tolerance(5m)    
  |eval(lambda: abs("historical.liters" - "actual.liters"))
.as('diff')
.keep()
  |eval(lambda: if("diff" > 100, "historical.liters", "actual.liters"))
.as('calibrated')
.keep()
  |eval(lambda: float("historical.liters"))
.as('last')
.keep()
  |eval(lambda: float("actual.liters"))
.as('actual')
.keep()
  |delete()
.field('historical.liters')
.field('actual.liters')
  |influxDBOut()
.create()
.database('testdata')
.retentionPolicy('autogen')
.measurement('overhead')

However when I run this over a replay using below:
kapacitor record query -query ‘SELECT “liters”,“wifi” FROM “testdata”.“autogen”.“overhead” WHERE time > now()-7d’ -type stream

I continuously get following error:
task start: cannot add child mismatched edges: mean5:stream -> join9:batch

Please bear in mind the data is being collected every 30 seconds. Kindly help.