fix: LPBatcher issue - lines longer than the buffer size lead to emitting 0 size packets and unemptied buffer

influxdb3/batching/lp_batcher.go

@@ -162,16 +162,30 @@ func (lpb *LPBatcher) Emit() []byte {
 }
 
 func (lpb *LPBatcher) emitBytes() []byte {
+	firstLF := bytes.IndexByte(lpb.buffer, '\n')
+
+	var packet []byte
+
 	c := min(lpb.size, len(lpb.buffer))
 
 	if c == 0 { // i.e. buffer is empty
 		return lpb.buffer
 	}
 
+	// With first line larger than defined size
+	// just emit first line
+	if firstLF > lpb.size {
+		packet = lpb.buffer[:firstLF]
+		lpb.buffer = lpb.buffer[len(packet)+1:] // remove trailing '\n'
+		return packet
+	}
+
+	// otherwise: process buffer where len(buffer) > size with multiple lines
+
 	prepacket := lpb.buffer[:c]
 	lastLF := bytes.LastIndexByte(prepacket, '\n') + 1
 
-	packet := lpb.buffer[:lastLF]
+	packet = lpb.buffer[:lastLF]
 	lpb.buffer = lpb.buffer[len(packet):]
 
 	return packet

influxdb3/batching/lp_batcher_test.go

@@ -283,3 +283,28 @@ func TestLPAddLargerThanSize(t *testing.T) {
 	assert.Equal(t, len(remainBuffer), lpb.CurrentLoadSize())
 	assert.Equal(t, remainBuffer, lpb.buffer)
 }
+
+// see EAR 5762
+func TestLPAddLinesLargerThanSize(t *testing.T) {
+	batchSize := 16
+	loadFactor := 10
+	capacity := batchSize * loadFactor
+
+	linesWithCRLF := []string{
+		"0123456789ABCDEFZZ", // len 18
+		"ZZFEDCBA9876543210", // len 18
+	}
+
+	emitCt := 0
+	resultBuffer := make([]byte, 0)
+	lpb := NewLPBatcher(
+		WithBufferSize(batchSize),
+		WithBufferCapacity(capacity),
+		WithEmitBytesCallback(func(ba []byte) {
+			emitCt++
+			resultBuffer = append(resultBuffer, ba...)
+		}))
+	lpb.Add(linesWithCRLF...)
+	assert.Equal(t, 2, emitCt, "Emit should be called correct number of times")
+	assert.Equal(t, strings.Join(linesWithCRLF, ""), string(resultBuffer))
+}