[pkg/stanza] Fix nil pointer dereference (#27882)

**Description:** 
Copying reader with empty FlushState will cause a nil pointer
dereference.

pkg/stanza/fileconsumer/internal/reader/factory.go

@@ -50,10 +50,7 @@ func (f *Factory) Copy(old *Reader, newFile *os.File) (*Reader, error) {
 		Offset:          old.Offset,
 		FileAttributes:  util.MapCopy(old.FileAttributes),
 		HeaderFinalized: old.HeaderFinalized,
-		FlushState: &flush.State{
-			LastDataChange: old.FlushState.LastDataChange,
-			LastDataLength: old.FlushState.LastDataLength,
-		},
+		FlushState:      old.FlushState.Copy(),
 	})
 }
 
@@ -71,12 +68,8 @@ func (f *Factory) build(file *os.File, m *Metadata) (r *Reader, err error) {
 		decoder:  decode.New(f.Encoding),
 	}
 
-	if m.FlushState == nil {
-		r.lineSplitFunc = trim.WithFunc(trim.ToLength(f.SplitFunc, f.Config.MaxLogSize), f.TrimFunc)
-	} else {
-		flushFunc := m.FlushState.Func(f.SplitFunc, f.Config.FlushTimeout)
-		r.lineSplitFunc = trim.WithFunc(trim.ToLength(flushFunc, f.Config.MaxLogSize), f.TrimFunc)
-	}
+	flushFunc := m.FlushState.Func(f.SplitFunc, f.Config.FlushTimeout)
+	r.lineSplitFunc = trim.WithFunc(trim.ToLength(flushFunc, f.Config.MaxLogSize), f.TrimFunc)
 
 	if !f.FromBeginning {
 		if err = r.offsetToEnd(); err != nil {

pkg/stanza/fileconsumer/reader_test.go

@@ -22,6 +22,21 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )
 
+func TestCopyReaderWithoutFlusher(t *testing.T) {
+	f, _ := testReaderFactory(t, split.Config{}, defaultMaxLogSize, 0)
+
+	temp := openTemp(t, t.TempDir())
+	fp, err := f.NewFingerprint(temp)
+	require.NoError(t, err)
+
+	r, err := f.NewReader(temp, fp)
+	require.NoError(t, err)
+
+	// A copy of the reader should not panic
+	_, err = f.Copy(r, temp)
+	assert.NoError(t, err)
+}
+
 func TestPersistFlusher(t *testing.T) {
 	flushPeriod := 100 * time.Millisecond
 	f, emitChan := testReaderFactory(t, split.Config{}, defaultMaxLogSize, flushPeriod)

pkg/stanza/flush/flush.go

@@ -13,11 +13,21 @@ type State struct {
 	LastDataLength int
 }
 
+func (s *State) Copy() *State {
+	if s == nil {
+		return nil
+	}
+	return &State{
+		LastDataChange: s.LastDataChange,
+		LastDataLength: s.LastDataLength,
+	}
+}
+
 // Func wraps a bufio.SplitFunc with a timer.
 // When the timer expires, an incomplete token may be returned.
 // The timer will reset any time the data parameter changes.
 func (s *State) Func(splitFunc bufio.SplitFunc, period time.Duration) bufio.SplitFunc {
-	if period <= 0 {
+	if s == nil || period <= 0 {
 		return splitFunc
 	}