fix(file): checkInterval validation (#2156)

Fix checkInterval validation and doc

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

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

@@ -69,6 +69,7 @@ every 5 seconds.
     {
       "file": {
         "path": "/tmp/result.txt",
+        "checkInterval": 5000,
         "fileType": "lines",
         "format": "json"
       }

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

@@ -13,7 +13,7 @@
 | checkInterval      | 是    | 定义 [rolling 策略](#rolling-策略)的属性之一。检查基于时间的滚动策略的间隔（以毫秒为单位），用于控制检查文件是否应该翻转的频率。    |
 | rollingCount       | 是    | 定义 [rolling 策略](#rolling-策略)的属性之一。文件翻转前的最大消息计数。                                |
 | rollingNamePattern | 是    | 定义 [rolling 策略](#rolling-策略)的属性之一。指定滚动文件创建时如何放置时间戳。时间戳可为“前缀”，“后缀”或“无”。         |
-| compression        | 是   | 使用指定的压缩方法压缩 Payload。当前支持 gzip, zstd 算法。                                       |
+| compression        | 是    | 使用指定的压缩方法压缩 Payload。当前支持 gzip, zstd 算法。                                        |
 
 其他通用的 sink 属性也支持，请参阅[公共属性](../overview.md#公共属性)。其中，`format` 属性用于定义文件中数据的格式。某些文件类型只能与特定格式一起使用，详情请参阅[文件类型](#文件类型)。
 
@@ -46,7 +46,7 @@
     {
       "file": {
         "path": "/tmp/result.txt",
-        "interval": 5000,
+        "checkInterval": 5000,
         "fileType": "lines",
         "format": "json"
       }

internal/io/file/file_sink.go

@@ -33,8 +33,8 @@ type sinkConf struct {
 	RollingInterval    int64    `json:"rollingInterval"`
 	RollingCount       int      `json:"rollingCount"`
 	RollingNamePattern string   `json:"rollingNamePattern"` // where to add the timestamp to the file name
-	CheckInterval      *int64   `json:"checkInterval"`      // Once interval removed, this will be NOT nullable
-	Path               string   `json:"path"`               // support dynamic property, when rolling, make sure the path is updated
+	CheckInterval      int64    `json:"checkInterval"`
+	Path               string   `json:"path"` // support dynamic property, when rolling, make sure the path is updated
 	FileType           FileType `json:"fileType"`
 	HasHeader          bool     `json:"hasHeader"`
 	Delimiter          string   `json:"delimiter"`
@@ -52,17 +52,14 @@ type fileSink struct {
 
 func (m *fileSink) Configure(props map[string]interface{}) error {
 	c := &sinkConf{
-		RollingCount: 1000000,
-		Path:         "cache",
-		FileType:     LINES_TYPE,
+		RollingCount:  1000000,
+		Path:          "cache",
+		FileType:      LINES_TYPE,
+		CheckInterval: (5 * time.Minute).Milliseconds(),
 	}
 	if err := cast.MapToStruct(props, c); err != nil {
 		return err
 	}
-	if c.CheckInterval == nil { // set checkInterval default value if both interval and checkInerval are not set
-		t := (5 * time.Minute).Milliseconds()
-		c.CheckInterval = &t
-	}
 	if c.RollingInterval < 0 {
 		return fmt.Errorf("rollingInterval must be positive")
 	}
@@ -70,7 +67,7 @@ func (m *fileSink) Configure(props map[string]interface{}) error {
 		return fmt.Errorf("rollingCount must be positive")
 	}
 
-	if *c.CheckInterval < 0 {
+	if c.CheckInterval < 0 {
 		return fmt.Errorf("checkInterval must be positive")
 	}
 	if c.RollingInterval == 0 && c.RollingCount == 0 {
@@ -107,8 +104,8 @@ func (m *fileSink) Configure(props map[string]interface{}) error {
 func (m *fileSink) Open(ctx api.StreamContext) error {
 	ctx.GetLogger().Debug("Opening file sink")
 	// Check if the files have opened longer than the rolling interval, if so close it and create a new one
-	if *m.c.CheckInterval > 0 {
-		t := conf.GetTicker(*m.c.CheckInterval)
+	if m.c.CheckInterval > 0 {
+		t := conf.GetTicker(m.c.CheckInterval)
 		go func() {
 			defer t.Stop()
 			for {

internal/io/file/file_sink_test.go

@@ -125,11 +125,7 @@ func TestConfigure(t *testing.T) {
 }
 
 func TestFileSink_Configure(t *testing.T) {
-	var (
-		defaultCheckInterval = (5 * time.Minute).Milliseconds()
-		int500               = 500
-		int64_500            = int64(int500)
-	)
+	defaultCheckInterval := (5 * time.Minute).Milliseconds()
 
 	tests := []struct {
 		name string
@@ -139,7 +135,7 @@ func TestFileSink_Configure(t *testing.T) {
 		{
 			name: "default configurations",
 			c: &sinkConf{
-				CheckInterval: &defaultCheckInterval,
+				CheckInterval: defaultCheckInterval,
 				Path:          "cache",
 				FileType:      LINES_TYPE,
 				RollingCount:  1000000,
@@ -149,7 +145,7 @@ func TestFileSink_Configure(t *testing.T) {
 		{
 			name: "new props",
 			c: &sinkConf{
-				CheckInterval:      &int64_500,
+				CheckInterval:      500,
 				Path:               "test",
 				FileType:           CSV_TYPE,
 				Format:             message.FormatDelimited,
@@ -168,7 +164,7 @@ func TestFileSink_Configure(t *testing.T) {
 		{ // only set rolling interval
 			name: "rolling",
 			c: &sinkConf{
-				CheckInterval:   &defaultCheckInterval,
+				CheckInterval:   defaultCheckInterval,
 				Path:            "cache",
 				FileType:        LINES_TYPE,
 				RollingInterval: 500,
@@ -182,7 +178,7 @@ func TestFileSink_Configure(t *testing.T) {
 		{
 			name: "fields",
 			c: &sinkConf{
-				CheckInterval:   &defaultCheckInterval,
+				CheckInterval:   defaultCheckInterval,
 				Path:            "cache",
 				FileType:        LINES_TYPE,
 				RollingInterval: 500,