refactor: remove deprecated properties (#1937)

* refactor(sink/runAsync): rm deprecated property

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

* refactor(filesink/interval): rm deprecated property

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

* fix: ci test failure

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

---------

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

docs/en_US/api/cli/data.md

@@ -28,7 +28,7 @@ The file format for importing and exporting Data is JSON, which can contain : `s
     "mqtt":"{\"td\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://broker.emqx.io:1883\",\"username\":\"admin\"},\"test\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://127.0.0.1:1883\",\"username\":\"admin\"}}"
   },
   "sinkConfig":{
-    "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"runAsync\":false,\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
+    "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
   },
   "connectionConfig":{
   },

docs/en_US/api/restapi/data.md

@@ -28,7 +28,7 @@ The file format for importing and exporting data is JSON, which can contain : `s
       "mqtt":"{\"td\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://broker.emqx.io:1883\",\"username\":\"admin\"},\"test\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://127.0.0.1:1883\",\"username\":\"admin\"}}"
     },
     "sinkConfig":{
-      "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"runAsync\":false,\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
+      "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
     },
     "connectionConfig":{
     },

docs/en_US/guide/sinks/builtin/file.md

@@ -14,7 +14,6 @@ connector that can read the file sink’s output.
 | checkInterval         | true     | One of the property to set the [rolling strategy](#rolling-strategy). The interval in millisecond for checking time based rolling policies. This controls the frequency to check whether a part file should rollover.                                              |
 | rollingCount          | true     | One of the property to set the [rolling strategy](#rolling-strategy). The maximum message counts in a file before rollover.                                                                                                                                        |
 | rollingNamePattern    | true     | One of the property to set the [rolling strategy](#rolling-strategy). Define how to named the rolling files by specifying where to put the timestamp during file creation. The value could be "prefix", "suffix" or "none".                                        |
-| interval (deprecated) | true     | This property is deprecated since 1.10 and will be removed later, please use checkInterval instead. The time interval (ms) for flushing the analysis result into the file. The default value is 1000, which means write the analysis result with every one second. |
 | compression           | true     | Compress the payload with the specified compression method. Support  `gzip`, `zstd` method now.                                                                                                                                                                    |
 
 Other common sink properties are supported. Please refer to

docs/zh_CN/api/cli/data.md

@@ -28,7 +28,7 @@ eKuiper 命令行工具允许您导入导出当前数据。
     "mqtt":"{\"td\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://broker.emqx.io:1883\",\"username\":\"admin\"},\"test\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://127.0.0.1:1883\",\"username\":\"admin\"}}"
   },
   "sinkConfig":{
-    "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"runAsync\":false,\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
+    "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
   },
   "connectionConfig":{
   },

docs/zh_CN/api/restapi/data.md

@@ -28,7 +28,7 @@ eKuiper REST api 允许您导入导出数据。
       "mqtt":"{\"td\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://broker.emqx.io:1883\",\"username\":\"admin\"},\"test\":{\"insecureSkipVerify\":false,\"password\":\"public\",\"protocolVersion\":\"3.1.1\",\"qos\":1,\"server\":\"tcp://127.0.0.1:1883\",\"username\":\"admin\"}}"
     },
     "sinkConfig":{
-      "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"runAsync\":false,\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
+      "edgex":"{\"test\":{\"bufferLength\":1024,\"contentType\":\"application/json\",\"enableCache\":false,\"format\":\"json\",\"messageType\":\"event\",\"omitIfEmpty\":false,\"port\":6379,\"protocol\":\"redis\",\"sendSingle\":true,\"server\":\"localhost\",\"topic\":\"application\",\"type\":\"redis\"}}"
     },
     "connectionConfig":{
     },

docs/zh_CN/guide/sinks/builtin/file.md

@@ -13,7 +13,6 @@
 | checkInterval      | 是    | 定义 [rolling 策略](#rolling-策略)的属性之一。检查基于时间的滚动策略的间隔（以毫秒为单位），用于控制检查文件是否应该翻转的频率。    |
 | rollingCount       | 是    | 定义 [rolling 策略](#rolling-策略)的属性之一。文件翻转前的最大消息计数。                                |
 | rollingNamePattern | 是    | 定义 [rolling 策略](#rolling-策略)的属性之一。指定滚动文件创建时如何放置时间戳。时间戳可为“前缀”，“后缀”或“无”。         |
-| interval (已弃用)     | 是    | 写入分析结果的时间间隔（毫秒）。 默认值为1000，这表示每隔一秒钟写入一次分析结果。                                    |
 | compression        | 	是   | 	使用指定的压缩方法压缩 Payload。当前支持 gzip, zstd 算法。                                       |
 
 其他通用的 sink 属性也支持，请参阅[公共属性](../overview.md#公共属性)。其中，`format` 属性用于定义文件中数据的格式。某些文件类型只能与特定格式一起使用，详情请参阅[文件类型](#文件类型)。

etc/sinks/file.json

@@ -127,21 +127,7 @@
 				"en_US": "Rolling Name Pattern",
 				"zh_CN": "Rolling 文件名模式"
 			}
-		}, {
-			"name": "interval (deprecated)",
-			"default": 1000,
-			"optional": true,
-			"control": "text",
-			"type": "int",
-			"hint": {
-				"en_US": "The time interval (ms) for writing the analysis result.",
-				"zh_CN": "写入分析结果的时间间隔（毫秒）。"
-			},
-			"label": {
-				"en_US": "Intervals",
-				"zh_CN": "间隔时间"
-			}
-	}],
+		}],
 	"node": {
 		"category": "sink",
 		"icon": "iconPath",

internal/io/file/file_sink.go

@@ -30,7 +30,6 @@ import (
 )
 
 type sinkConf struct {
-	Interval           *int     `json:"interval"` // deprecated, will remove in the next release
 	RollingInterval    int64    `json:"rollingInterval"`
 	RollingCount       int      `json:"rollingCount"`
 	RollingNamePattern string   `json:"rollingNamePattern"` // where to add the timestamp to the file name
@@ -60,17 +59,7 @@ func (m *fileSink) Configure(props map[string]interface{}) error {
 	if err := cast.MapToStruct(props, c); err != nil {
 		return err
 	}
-	if c.Interval != nil {
-		if *c.Interval < 0 {
-			return fmt.Errorf("interval must be positive")
-		} else if c.CheckInterval == nil {
-			conf.Log.Warnf("interval is deprecated, use checkInterval instead. automatically set checkInterval to %d", c.Interval)
-			t := int64(*c.Interval)
-			c.CheckInterval = &t
-		} else {
-			conf.Log.Warnf("interval is deprecated and ignored, use checkInterval instead.")
-		}
-	} else if c.CheckInterval == nil { // set checkInterval default value if both interval and checkInerval are not set
+	if c.CheckInterval == nil { // set checkInterval default value if both interval and checkInerval are not set
 		t := (5 * time.Minute).Milliseconds()
 		c.CheckInterval = &t
 	}

internal/io/file/file_sink_test.go

@@ -42,16 +42,9 @@ func TestConfigure(t *testing.T) {
 	if err != nil {
 		t.Errorf("Configure() error = %v, wantErr nil", err)
 	}
-	if *m.c.Interval != 500 {
-		t.Errorf("Configure() Interval = %v, want 500", m.c.Interval)
-	}
 	if m.c.Path != "test" {
 		t.Errorf("Configure() Path = %v, want test", m.c.Path)
 	}
-	err = m.Configure(map[string]interface{}{"interval": -1, "path": "test"})
-	if err == nil {
-		t.Errorf("Configure() error = %v, wantErr not nil", err)
-	}
 	err = m.Configure(map[string]interface{}{"interval": 500, "path": ""})
 	if err == nil {
 		t.Errorf("Configure() error = %v, wantErr not nil", err)
@@ -154,21 +147,6 @@ func TestFileSink_Configure(t *testing.T) {
 			p: map[string]interface{}{},
 		},
 		{
-			name: "previous setting",
-			c: &sinkConf{
-				Interval:      &int500,
-				CheckInterval: &int64_500,
-				Path:          "test",
-				FileType:      LINES_TYPE,
-				RollingCount:  1000000,
-			},
-
-			p: map[string]interface{}{
-				"interval": 500,
-				"path":     "test",
-			},
-		},
-		{
 			name: "new props",
 			c: &sinkConf{
 				CheckInterval:      &int64_500,

internal/topo/node/sink_node.go

@@ -37,7 +37,6 @@ import (
 
 type SinkConf struct {
 	Concurrency    int      `json:"concurrency"`
-	RunAsync       bool     `json:"runAsync"` // deprecated, will remove in the next release
 	Omitempty      bool     `json:"omitIfEmpty"`
 	SendSingle     bool     `json:"sendSingle"`
 	DataTemplate   string   `json:"dataTemplate"`
@@ -187,9 +186,6 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 									}
 									stats.SetBufferLength(int64(len(m.input)))
 									stats.IncTotalRecordsIn()
-									if sconf.RunAsync {
-										conf.Log.Warnf("RunAsync is deprecated and ignored.")
-									}
 									err := doCollect(ctx, sink, data, sendManager, stats, sconf)
 									if err != nil {
 										logger.Warnf("sink collect error: %v", err)
@@ -204,10 +200,7 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 							}
 						} else {
 							logger.Infof("Creating sink cache")
-							if sconf.RunAsync { // async mode, the ack must have an id
-								// is not supported and validated in the configure, should not go here
-								return fmt.Errorf("async mode is not supported for cache sink")
-							} else { // sync mode, the ack is already in order
+							{ // sync mode, the ack is already in order
 								dataCh := make(chan []map[string]interface{}, sconf.BufferLength)
 								c := cache.NewSyncCache(ctx, dataCh, result, stats, &sconf.SinkConf, sconf.BufferLength)
 								for {
@@ -256,7 +249,6 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 									}
 								}
 							}
-
 						}
 					})
 					if panicOrError != nil {
@@ -275,7 +267,6 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 func (m *SinkNode) parseConf(logger api.Logger) (*SinkConf, error) {
 	sconf := &SinkConf{
 		Concurrency:  1,
-		RunAsync:     false,
 		Omitempty:    false,
 		SendSingle:   false,
 		DataTemplate: "",
@@ -301,10 +292,6 @@ func (m *SinkNode) parseConf(logger api.Logger) (*SinkConf, error) {
 	if err != nil {
 		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")
-	}
 	if sconf.DataField == "" {
 		if v, ok := m.options["tableDataField"]; ok {
 			sconf.DataField = v.(string)

internal/topo/node/sink_node_test.go

@@ -417,16 +417,6 @@ func TestConfig(t *testing.T) {
 		}, {
 			config: map[string]interface{}{
 				"enableCache":          true,
-				"memoryCacheThreshold": 2,
-				"bufferPageSize":       2,
-				"runAsync":             true,
-				"maxDiskCache":         6,
-				"resendInterval":       10,
-			},
-			err: errors.New("cache is not supported for async sink, do not use enableCache and runAsync properties together"),
-		}, {
-			config: map[string]interface{}{
-				"enableCache":          true,
 				"memoryCacheThreshold": 256,
 				"bufferLength":         10,
 				"maxDiskCache":         6,