chore: deprecate sink runAsync property (#1809)

* deprecate runasync

Signed-off-by: Rui-Gan <1171530954@qq.com>

* increase test coverage

Signed-off-by: Rui-Gan <1171530954@qq.com>

---------

Signed-off-by: Rui-Gan <1171530954@qq.com>

docs/zh_CN/guide/sinks/overview.md

@@ -69,7 +69,7 @@
 |----------------------|------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | concurrency          | int: 1                             | 设置运行的线程数。该参数值大于1时，消息发出的顺序可能无法保证。                                                                                                                                                                                                                 |
 | bufferLength         | int: 1024                          | 设置可缓存消息数目。若缓存消息数超过此限制，sink将阻塞消息接收，直到缓存消息被消费使得缓存消息数目小于限制为止。                                                                                                                                                                                       |
-| runAsync             | bool:false                         | 设置是否异步运行输出操作以提升性能。请注意，异步运行的情况下，输出结果顺序不能保证。                                                                                                                                                                                                       |
+| runAsync(deprecated)             | bool:false                         | 设置是否异步运行输出操作以提升性能。请注意，异步运行的情况下，输出结果顺序不能保证。                                                                                                                                                                                                       |
 | omitIfEmpty          | bool: false                        | 如果配置项设置为 true，则当 SELECT 结果为空时，该结果将不提供给目标运算符。                                                                                                                                                                                                     |
 | sendSingle           | bool: false                        | 输出消息以数组形式接收，该属性意味着是否将结果一一发送。 如果为false，则输出消息将为`{"result":"${the string of received message}"}`。 例如，`{"result":"[{\"count\":30},"\"count\":20}]"}`。否则，结果消息将与实际字段名称一一对应发送。 对于与上述相同的示例，它将发送 `{"count":30}`，然后发送`{"count":20}`到 RESTful 端点。默认为 false。 |
 | dataTemplate         | string: ""                         | [golang 模板](https://golang.org/pkg/html/template)格式字符串，用于指定输出数据格式。 模板的输入是目标消息，该消息始终是映射数组。 如果未指定数据模板，则将数据作为原始输入。                                                                                                                                  |

internal/topo/node/sink_node.go

@@ -16,6 +16,9 @@ package node
 
 import (
 	"fmt"
+	"strings"
+	"sync"
+
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
@@ -29,13 +32,11 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 	"github.com/lf-edge/ekuiper/pkg/message"
-	"strings"
-	"sync"
 )
 
 type SinkConf struct {
 	Concurrency  int    `json:"concurrency"`
-	RunAsync     bool   `json:"runAsync"`
+	RunAsync     bool   `json:"runAsync"` // deprecated, will remove in the next release
 	Omitempty    bool   `json:"omitIfEmpty"`
 	SendSingle   bool   `json:"sendSingle"`
 	DataTemplate string `json:"dataTemplate"`
@@ -165,23 +166,11 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 									stats.SetBufferLength(int64(len(m.input)))
 									stats.IncTotalRecordsIn()
 									if sconf.RunAsync {
-										go func() {
-											p := infra.SafeRun(func() error {
-												err := doCollect(ctx, sink, data, stats, sconf)
-												if err != nil {
-													logger.Warnf("sink collect error: %v", err)
-												}
-												return nil
-											})
-											if p != nil {
-												infra.DrainError(ctx, p, result)
-											}
-										}()
-									} else {
-										err := doCollect(ctx, sink, data, stats, sconf)
-										if err != nil {
-											logger.Warnf("sink collect error: %v", err)
-										}
+										conf.Log.Warnf("RunAsync is deprecated and ignored.")
+									}
+									err := doCollect(ctx, sink, data, stats, sconf)
+									if err != nil {
+										logger.Warnf("sink collect error: %v", err)
 									}
 								case <-ctx.Done():
 									logger.Infof("sink node %s instance %d done", m.name, instance)
@@ -291,6 +280,7 @@ func (m *SinkNode) parseConf(logger api.Logger) (*SinkConf, error) {
 		return nil, fmt.Errorf("read properties %v to cache conf fail with error: %v", m.options, err)
 	}
 	if sconf.SinkConf.EnableCache && sconf.RunAsync {
+		conf.Log.Warnf("RunAsync is deprecated and ignored.")
 		return nil, fmt.Errorf("cache is not supported for async sink, do not use enableCache and runAsync properties together")
 	}
 	err = sconf.SinkConf.Validate()

internal/topo/node/sink_node_test.go

@@ -417,3 +417,76 @@ func TestConfig(t *testing.T) {
 		}
 	}
 }
+
+func TestSinkNode_reset(t *testing.T) {
+	mockSink := mocknode.NewMockSink()
+	s := NewSinkNodeWithSink("mockSink", mockSink, nil)
+	s.reset()
+	if s.statManagers != nil {
+		t.Errorf("reset() failed")
+	}
+}
+
+func Test_getSink(t *testing.T) {
+	_, err := getSink("mock", map[string]interface{}{"sendSingle": true, "omitIfEmpty": true})
+	if err == nil {
+		t.Errorf("getSink() failed")
+	}
+}
+
+func Test_itemToMap(t *testing.T) {
+	type args struct {
+		item interface{}
+	}
+	tests := []struct {
+		name string
+		args args
+		want []map[string]interface{}
+	}{
+		{
+			name: "test1",
+			args: args{
+				item: errors.New("test"),
+			},
+			want: []map[string]interface{}{
+				{"error": "test"},
+			},
+		},
+		{
+			name: "test2",
+			args: args{
+				item: "test2",
+			},
+			want: []map[string]interface{}{
+				{"error": fmt.Sprintf("result is not a map slice but found %#v", "test2")},
+			},
+		},
+		{
+			name: "test3",
+			args: args{
+				item: xsql.Row(&xsql.Tuple{Emitter: "a", Message: map[string]interface{}{"a": 1, "b": "2"}, Timestamp: conf.GetNowInMilli(), Metadata: nil}),
+			},
+			want: []map[string]interface{}{
+				{"a": 1, "b": "2"},
+			},
+		},
+		{
+			name: "test4",
+			args: args{
+				item: xsql.Collection(&xsql.WindowTuples{Content: []xsql.TupleRow{
+					&xsql.Tuple{Emitter: "a", Message: map[string]interface{}{"a": 1, "b": "2"}, Timestamp: conf.GetNowInMilli(), Metadata: nil},
+				}}),
+			},
+			want: []map[string]interface{}{
+				{"a": 1, "b": "2"},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := itemToMap(tt.args.item); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("itemToMap() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}