tasks stop working repeatedly after every a few days

Nobody has claimed this yet.

My kapacitor instance have two different influxdb servers, each of them have some tasks enabled, and every a few days, all tasks from one specific influxdb will stop working.

running `kapacitor show task` give no error, nothing weird in logs, if I run `kapacitor define task` to update the problem task, it will hang. The only way to recover is to restart kapacitor.

At first I thought this is similar to #1924 #1149 #1829
However my issue was more likely relate to data subscription since show tasks give me 0 batch throughput, see one of the problematic task below:

```
ID: as2s_idc_alert
Error:
Template:
Type: batch
Status: enabled
Executing: true
Created: 15 Aug 18 12:08 UTC
Modified: 20 Sep 18 08:57 UTC
LastEnabled: 20 Sep 18 08:57 UTC
Databases Retention Policies: ["quality_report"."7day"]
TICKscript:
dbrp "quality_report"."7day"

var data = batch
|query('''SELECT sum("recvNum")/sum("sendNum")*100 AS quality FROM "quality_report"."7day"."raw_as2s"''')
.period(5m)
.every(1m)
.groupBy(time(1s), 'srcIdc', 'dstIdc')
.align()
.cluster('influxdb_main')
|where(lambda: "quality" < 99.5)
|httpOut('debug')

var dataBySrcIdc = data
|groupBy('srcIdc')
|stateCount(lambda: "quality" < 98)
|httpOut('bysrc')
|alert()
.warn(lambda: "state_count" > 5 OR "quality" < 80)
.crit(lambda: "state_count" > 15 OR ("state_count" > 5 AND "quality" < 80))
.stateChangesOnly()
.message('as2s quality is bad in {{ index .Tags "srcIdc" }} with more than {{ index .Fields "state_count" }} dst idc quality below 98%, the worst quality is {{ index .Fields "quality" | printf "%0.1f"}}')
.alerta()
.resource('{{ .Name }}')
.event('{{ .ID }}')
.environment('Production')
.services('as2s')
.timeout(24h)
.postTimeout(10s)
.topic('media-as2s')

var dataByDstIdc = data
|groupBy('dstIdc')
|stateCount(lambda: "quality" < 98)
|httpOut('bydst')
|alert()
.warn(lambda: "state_count" > 5 OR "quality" < 80)
.crit(lambda: "state_count" > 15 OR ("state_count" > 5 AND "quality" < 80))
.message('as2s quality is bad in {{ index .Tags "dstIdc" }} with more than {{ index .Fields "state_count" }} src idc quality below 98%, the worst quality is {{ index .Fields "quality" | printf "%0.1f"}}')
.stateChangesOnly()
.alerta()
.resource('{{ .Name }}')
.event('{{ .ID }}')
.environment('Production')
.services('as2s')
.timeout(24h)
.postTimeout(10s)
.topic('media-as2s')

DOT:
digraph as2s_idc_alert {
graph [throughput="0.00 batches/s"];

query1 [avg_exec_time_ns="29.514690856s" batches_queried="31239246" errors="0" points_queried="116800778" working_cardinality="0" ];
query1 -> where2 [processed="31239246"];

where2 [avg_exec_time_ns="17.464µs" errors="0" working_cardinality="7932" ];
where2 -> http_out3 [processed="31239246"];

http_out3 [avg_exec_time_ns="14.454µs" errors="0" working_cardinality="7932" ];
http_out3 -> groupby8 [processed="31239246"];
http_out3 -> groupby4 [processed="31239246"];

groupby8 [avg_exec_time_ns="23.464µs" errors="0" working_cardinality="1" ];
groupby8 -> state_count9 [processed="691684"];

state_count9 [avg_exec_time_ns="2.52µs" errors="0" working_cardinality="94" ];
state_count9 -> http_out10 [processed="691684"];

http_out10 [avg_exec_time_ns="8.349µs" errors="0" working_cardinality="94" ];
http_out10 -> alert11 [processed="691684"];

alert11 [alerts_inhibited="0" alerts_triggered="6393" avg_exec_time_ns="6.424756ms" crits_triggered="35" errors="0" infos_triggered="0" oks_triggered="3181" warns_triggered="3177" working_cardinality="94" ];

groupby4 [avg_exec_time_ns="7.253µs" errors="0" working_cardinality="2" ];
groupby4 -> state_count5 [processed="1024139"];

state_count5 [avg_exec_time_ns="41.419µs" errors="0" working_cardinality="94" ];
state_count5 -> http_out6 [processed="1024139"];

http_out6 [avg_exec_time_ns="20.243µs" errors="0" working_cardinality="94" ];
http_out6 -> alert7 [processed="1024139"];

alert7 [alerts_inhibited="0" alerts_triggered="6990" avg_exec_time_ns="6.226751ms" crits_triggered="0" errors="0" infos_triggered="0" oks_triggered="3495" warns_triggered="3495" working_cardinality="94" ];
}
```

our kapacitor already suffer from this issue for months, I tried analyze goroutine dump but couldn't find out what goes wrong 😢
[kp_dump.txt](https://github.com/influxdata/kapacitor/files/2421144/kp_dump.txt)

Start with the reported sequence: tasks querying one InfluxDB server stop producing batches, `kapacitor show task` reports zero throughput, and `kapacitor define task` hangs until Kapacitor restarts. Read the attached `kp_dump.txt` goroutine dump and compare it with the task's batch subscription and `.cluster('influxdb_main')` configuration. Done means the failure mechanism is identified and tasks recover without restarting Kapacitor.

Written by the indexing model from the issue text.