Sink cache concurrent access problem large scale and long running rules

[ngjaying]

During testing large scale rules with 8000 rules running and sending 1 message per 10 second per rule for about 2 hours, Kuiper panics for concurrent sink cache save with the below message:

goroutine 53 [running]:
runtime.throw(0xd4b8a4, 0x21)
/usr/local/go/src/runtime/panic.go:774 +0x72 fp=0xc0b33c3388 sp=0xc0b33c3358 pc=0x6af7d2
runtime.mapaccess2(0xc70b80, 0xc02e178cf0, 0xc0a28babc0, 0xc0a28babc0, 0xc0b33c3470)
/usr/local/go/src/runtime/map.go:470 +0x278 fp=0xc0b33c33d0 sp=0xc0b33c3388 pc=0x68ef48
reflect.mapaccess(0xc70b80, 0xc02e178cf0, 0xc0a28babc0, 0xd4378c)
/usr/local/go/src/runtime/map.go:1319 +0x3f fp=0xc0b33c3408 sp=0xc0b33c33d0 pc=0x69125f
reflect.Value.MapIndex(0xc70b80, 0xc005fe3130, 0x195, 0xc4e5e0, 0xc0a28babc0, 0x82, 0x7fe8d0b54fff, 0x400, 0x7fe8d0a85d00)
/usr/local/go/src/reflect/value.go:1176 +0x16d fp=0xc0b33c3480 sp=0xc0b33c3408 pc=0x7132cd
encoding/gob.(*Encoder).encodeMap(0xc0b0ed4be0, 0xc0bd8a2cc0, 0xc70b80, 0xc005fe3130, 0x195, 0xd65260, 0xd65268, 0x0, 0x0)
/usr/local/go/src/encoding/gob/encode.go:375 +0x1e8 fp=0xc0b33c3520 sp=0xc0b33c3480 pc=0x9d6e18
encoding/gob.encOpFor.func3(0xc00016a0c0, 0xc093721bc0, 0xc70b80, 0xc005fe3130, 0x195)
/usr/local/go/src/encoding/gob/encode.go:571 +0xa7 fp=0xc0b33c35a0 sp=0xc0b33c3520 pc=0x9e1d87
encoding/gob.(*Encoder).encodeStruct(0xc0b0ed4be0, 0xc0bd8a2cc0, 0xc049e5f320, 0xcb1aa0, 0xc005fe3130, 0x199)
/usr/local/go/src/encoding/gob/encode.go:328 +0x268 fp=0xc0b33c3678 sp=0xc0b33c35a0 pc=0x9d6508
encoding/gob.(*Encoder).encode(0xc0b0ed4be0, 0xc0bd8a2cc0, 0xca84a0, 0xc005fe3130, 0x16, 0xc000145e80)
/usr/local/go/src/encoding/gob/encode.go:701 +0x16f fp=0xc0b33c3720 sp=0xc0b33c3678 pc=0x9da09f
encoding/gob.(*Encoder).encodeInterface(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc76040, 0xc0b6620400, 0x94)
/usr/local/go/src/encoding/gob/encode.go:419 +0x537 fp=0xc0b33c3860 sp=0xc0b33c3720 pc=0x9d7497
encoding/gob.encOpFor.func5(0xc000154540, 0xc093721b40, 0xc76040, 0xc0b6620400, 0x94)
/usr/local/go/src/encoding/gob/encode.go:589 +0x6a fp=0xc0b33c38a0 sp=0xc0b33c3860 pc=0x9e209a
encoding/gob.(*Encoder).encodeStruct(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc0000974e0, 0xcbec00, 0xc0b6620400, 0x99)
/usr/local/go/src/encoding/gob/encode.go:328 +0x268 fp=0xc0b33c3978 sp=0xc0b33c38a0 pc=0x9d6508
encoding/gob.encOpFor.func4(0x0, 0xc093721b00, 0xcbec00, 0xc0b6620400, 0x99)
/usr/local/go/src/encoding/gob/encode.go:581 +0xac fp=0xc0b33c39d0 sp=0xc0b33c3978 pc=0x9e1fac
encoding/gob.encodeReflectValue(0xc093721b00, 0xcbec00, 0xc0b6620400, 0x99, 0xc000130590, 0x0)
/usr/local/go/src/encoding/gob/encode.go:363 +0xcf fp=0xc0b33c3a50 sp=0xc0b33c39d0 pc=0x9d6b5f
encoding/gob.(*Encoder).encodeMap(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc70a00, 0xc000145e08, 0x195, 0xd65280, 0xc000130590, 0x0, 0x0)
/usr/local/go/src/encoding/gob/encode.go:375 +0x22e fp=0xc0b33c3af0 sp=0xc0b33c3a50 pc=0x9d6e5e
encoding/gob.encOpFor.func3(0xc0000fddd0, 0xc093721ac0, 0xc70a00, 0xc000145e08, 0x195)
/usr/local/go/src/encoding/gob/encode.go:571 +0xa7 fp=0xc0b33c3b70 sp=0xc0b33c3af0 pc=0x9e1d87
encoding/gob.(*Encoder).encodeSingle(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc0000974c0, 0xc70a00, 0xc000145e08, 0x195)
/usr/local/go/src/encoding/gob/encode.go:301 +0x1c5 fp=0xc0b33c3c08 sp=0xc0b33c3b70 pc=0x9d61b5
encoding/gob.(*Encoder).encode(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc2e820, 0xc000145e08, 0x16, 0xc000144180)
/usr/local/go/src/encoding/gob/encode.go:703 +0x1b7 fp=0xc0b33c3cb0 sp=0xc0b33c3c08 pc=0x9da0e7
encoding/gob.(*Encoder).EncodeValue(0xc0b0ed4be0, 0xc2e820, 0xc000145e08, 0x16, 0x0, 0x0)
/usr/local/go/src/encoding/gob/encoder.go:251 +0x3ce fp=0xc0b33c3d88 sp=0xc0b33c3cb0 pc=0x9db77e
encoding/gob.(*Encoder).Encode(0xc0b0ed4be0, 0xc2e820, 0xc000145e08, 0xc78740, 0xc0b659bec0)
/usr/local/go/src/encoding/gob/encoder.go:176 +0xa4 fp=0xc0b33c3dd8 sp=0xc0b33c3d88 pc=0x9db194
github.com/patrickmn/go-cache.(*cache).Save(0xc000145e00, 0xe0bc00, 0xc0b1be4f48, 0x0, 0x0)
/home/ubuntu/go/pkg/mod/github.com/patrickmn/go-cache@v2.1.0+incompatible/cache.go:975 +0x1aa fp=0xc0b33c3ee8 sp=0xc0b33c3dd8 pc=0x9ecf9a
github.com/patrickmn/go-cache.(*cache).SaveFile(0xc000145e00, 0xc0000790e0, 0x5c, 0x0, 0x0)
/home/ubuntu/go/pkg/mod/github.com/patrickmn/go-cache@v2.1.0+incompatible/cache.go:989 +0x86 fp=0xc0b33c3f48 sp=0xc0b33c3ee8 pc=0x9ed0b6
github.com/emqx/kuiper/common.(*SimpleKVStore).run(0xc000145e40)
/home/ubuntu/kuiper/common/kv.go:111 +0xe2 fp=0xc0b33c3fd8 sp=0xc0b33c3f48 pc=0xa21272
runtime.goexit()
/usr/local/go/src/runtime/asm_amd64.s:1357 +0x1 fp=0xc0b33c3fe0 sp=0xc0b33c3fd8 pc=0x6dccb1
created by github.com/emqx/kuiper/common.NewSimpleKVStore
/home/ubuntu/kuiper/common/kv.go:74 +0xd3