Use SourceID when spread events

pipeline/pipeline.go

@@ -471,7 +471,7 @@ func (p *Pipeline) streamEvent(event *Event) uint64 {
 
 	// spread events across all processors
 	if p.useSpread {
-		streamID = StreamID(event.SeqID % uint64(p.procCount.Load()))
+		streamID = StreamID(uint64(event.SourceID) % uint64(p.procCount.Load()))
 	}
 
 	if !p.disableStreams {

pipeline/pipeline_test.go

@@ -1,11 +1,14 @@
 package pipeline_test
 
 import (
+	"math/rand"
 	"reflect"
+	"sync"
 	"testing"
 
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/plugin/input/fake"
+	"github.com/ozontech/file.d/test"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 )
@@ -121,3 +124,38 @@ func TestInInvalidMessages(t *testing.T) {
 		})
 	}
 }
+
+func TestPipelineSequenceWhenUseSpread(t *testing.T) {
+	t.Parallel()
+
+	p, input, output := test.NewPipelineMock(nil, "passive", "parallel")
+	p.UseSpread()
+	p.Start()
+
+	const iters = 1000
+
+	wg := &sync.WaitGroup{}
+	wg.Add(iters)
+
+	outEvents := make([]string, 0)
+	mu := sync.Mutex{}
+	offsetSequence := int64(1)
+	output.SetOutFn(func(event *pipeline.Event) {
+		outEvents = append(outEvents, event.Root.EncodeToString())
+		mu.Lock()
+		defer mu.Unlock()
+
+		require.Equal(t, offsetSequence, event.Offset)
+		offsetSequence++
+
+		wg.Done()
+	})
+
+	sourceID := pipeline.SourceID(rand.Int())
+	for offset := int64(1); offset <= iters; offset++ {
+		input.In(sourceID, "", offset, []byte(`{}`))
+	}
+
+	wg.Wait()
+	p.Stop()
+}

pipeline/pipeline_whitebox_test.go

@@ -31,7 +31,7 @@ func TestPipeline_streamEvent(t *testing.T) {
 	p.UseSpread()
 	p.streamEvent(event)
 
-	expectedStreamID := StreamID(event.SeqID % uint64(procs))
+	expectedStreamID := StreamID(uint64(event.SourceID) % uint64(procs))
 
 	assert.Equal(t, event, p.streamer.getStream(expectedStreamID, DefaultStreamName).first)
 }