Fix stuck

e2e/split_join/split_join.go

@@ -7,7 +7,6 @@ import (
 	"path"
 	"path/filepath"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -80,31 +79,26 @@ func (c *Config) Send(t *testing.T) {
 func (c *Config) Validate(t *testing.T) {
 	r := require.New(t)
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
 	defer cancel()
 
 	expectedEventsCount := messages * arrayLen
 
 	strBuilder := strings.Builder{}
 	gotEvents := 0
-	wg := sync.WaitGroup{}
-	wg.Add(expectedEventsCount)
+	done := make(chan struct{})
 
 	go func() {
 		r.NoError(c.consumer.Consume(ctx, []string{c.topic}, handlerFunc(func(msg *sarama.ConsumerMessage) {
 			strBuilder.Write(msg.Value)
 			strBuilder.WriteString("\n")
 			gotEvents++
-			wg.Done()
+			if gotEvents == expectedEventsCount {
+				close(done)
+			}
 		})))
 	}()
 
-	done := make(chan struct{})
-	go func() {
-		wg.Wait()
-		close(done)
-	}()
-
 	select {
 	case <-done:
 	case <-ctx.Done():

pipeline/batch.go

@@ -77,7 +77,7 @@ func (b *Batch) reset() {
 }
 
 func (b *Batch) append(e *Event) {
-	if e.IsChildParentKind() {
+	if !e.IsChildKind() {
 		b.hasIterableEvents = true
 	}
 
@@ -91,10 +91,7 @@ func (b *Batch) updateStatus() BatchStatus {
 		// batch is empty
 		return BatchStatusNotReady
 	}
-	if !b.hasIterableEvents {
-		// batch contains only parents
-		return BatchStatusNotReady
-	}
+
 	switch {
 	case (b.maxSizeCount != 0 && l >= b.maxSizeCount) || (b.maxSizeBytes != 0 && b.maxSizeBytes <= b.eventsSize):
 		b.status = BatchStatusMaxSizeExceeded