Fix after merge build errors

e2e/split_join/config.yml

@@ -4,9 +4,6 @@ pipelines:
       event_timeout: 1h
     input:
       type: file
-      persistence_mode: async
-      watching_dir: SOME_DIR
-      offsets_file: SOME_FILE
       offsets_op: reset
     actions:
       - type: split

e2e/split_join/remap_join.go → e2e/split_join/split_join.go

@@ -37,7 +37,9 @@ func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string)
 func (c *Config) Send(t *testing.T) {
 	file, err := os.Create(path.Join(c.inputDir, "input.log"))
 	require.NoError(t, err)
-	defer file.Close()
+	defer func(file *os.File) {
+		_ = file.Close()
+	}(file)
 
 	for i := 0; i < c.count; i++ {
 		_, err = file.WriteString(`{ "data": [ { "hello": "world" }, { "message": "start " }, { "message": "continue" }, { "file": ".d" }, { "open": "source" } ] }` + "\n")

pipeline/pipeline.go

@@ -505,7 +505,7 @@ func (p *Pipeline) Error(err string) {
 }
 
 func (p *Pipeline) finalize(event *Event, notifyInput bool, backEvent bool) {
-	if event.IsTimeoutKind() {
+	if event.IsTimeoutKind() || event.IsChildKind() {
 		return
 	}
 

pipeline/processor.go

@@ -352,16 +352,16 @@ func (p *processor) Propagate(event *Event) {
 // Any attempts to ActionHold or ActionCollapse the event will be suppressed by timeout events.
 func (p *processor) Spawn(parent *Event, nodes []*insaneJSON.Node) {
 	parent.SetChildParentKind()
-	nextActionIdx := parent.action.Load() + 1
+	nextActionIdx := parent.action + 1
 
 	for _, node := range nodes {
 		child := newEvent()
 		parent.children = append(parent.children, child)
 		child.Root.MutateToNode(node)
 		child.SetChildKind()
-		child.action.Store(nextActionIdx)
+		child.action = nextActionIdx
 
-		ok := p.doActions(child)
+		ok, _ := p.doActions(child)
 		if ok {
 			child.stage = eventStageOutput
 			p.output.Out(child)
@@ -378,7 +378,7 @@ func (p *processor) Spawn(parent *Event, nodes []*insaneJSON.Node) {
 		}
 
 		timeout := newTimeoutEvent(parent.stream)
-		timeout.action.Store(int64(i))
+		timeout.action = i
 		p.doActions(timeout)
 	}
 }

plugin/action/split/split.go

@@ -4,7 +4,6 @@ import (
 	"github.com/ozontech/file.d/cfg"
 	"github.com/ozontech/file.d/fd"
 	"github.com/ozontech/file.d/pipeline"
-	"github.com/ozontech/file.d/plugin"
 	insaneJSON "github.com/vitkovskii/insane-json"
 	"go.uber.org/zap"
 )
@@ -38,7 +37,6 @@ type Plugin struct {
 	config           *Config
 	logger           *zap.Logger
 	pluginController pipeline.ActionPluginController
-	plugin.NoMetricsPlugin
 }
 
 // ! config-params

plugin/output/clickhouse/clickhouse.go

@@ -412,7 +412,8 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) {
 	data := (*workerData).(data)
 	data.reset()
 
-	for _, event := range batch.Events {
+	for batch.Next() {
+		event := batch.Value()
 		for _, col := range data.cols {
 			node := event.Root.Dig(col.Name)