Remove references to Batch.FreeEntries

Remove all references, usages and tests to `Batch.FreeEntries` that
was used only by the shipper.

libbeat/outputs/elasticsearch/client_test.go

@@ -70,9 +70,8 @@ func (bm *batchMock) ACK() {
 func (bm *batchMock) Drop() {
 	bm.drop = true
 }
-func (bm *batchMock) Retry()       { panic("unimplemented") }
-func (bm *batchMock) Cancelled()   { panic("unimplemented") }
-func (bm *batchMock) FreeEntries() {}
+func (bm *batchMock) Retry()     { panic("unimplemented") }
+func (bm *batchMock) Cancelled() { panic("unimplemented") }
 func (bm *batchMock) SplitRetry() bool {
 	if bm.canSplit {
 		bm.didSplit = true

libbeat/outputs/outest/batch.go

@@ -77,8 +77,6 @@ func (b *Batch) SplitRetry() bool {
 	return len(b.events) > 1
 }
 
-func (b *Batch) FreeEntries() {}
-
 func (b *Batch) Cancelled() {
 	b.doSignal(BatchSignal{Tag: BatchCancelled})
 }

libbeat/publisher/event.go

@@ -47,17 +47,6 @@ type Batch interface {
 	// batch.Drop() if necessary).
 	SplitRetry() bool
 
-	// Release the internal pointer to this batch's events but do not yet
-	// acknowledge this batch. This exists specifically for the shipper output,
-	// where there is potentially a long gap between when events are handed off
-	// to the shipper and when they are acknowledged upstream; during that time,
-	// we need to preserve batch metadata for producer end-to-end acknowledgments,
-	// but we do not need the events themselves since they are already queued by
-	// the shipper. It is only guaranteed to release event pointers when using the
-	// proxy queue.
-	// Never call this on a batch that might be retried.
-	FreeEntries()
-
 	// Send was aborted, try again but don't decrease the batch's TTL counter.
 	Cancelled()
 }

libbeat/publisher/pipeline/ttl_batch.go

@@ -77,7 +77,6 @@ func newBatch(retryer retryer, original queue.Batch, ttl int) *ttlBatch {
 			events = append(events, event)
 		}
 	}
-	original.FreeEntries()
 
 	b := &ttlBatch{
 		done:    original.Done,
@@ -166,10 +165,6 @@ func (b *ttlBatch) RetryEvents(events []publisher.Event) {
 	b.Retry()
 }
 
-func (b *ttlBatch) FreeEntries() {
-	b.events = nil
-}
-
 // reduceTTL reduces the time to live for all events that have no 'guaranteed'
 // sending requirements.  reduceTTL returns true if the batch is still alive.
 func (b *ttlBatch) reduceTTL() bool {

libbeat/publisher/pipeline/ttl_batch_test.go

@@ -112,12 +112,6 @@ func TestBatchCallsDoneAndFreesEvents(t *testing.T) {
 	require.True(t, doneCalled, "Calling batch.Drop should invoke the done callback")
 }
 
-func TestNewBatchFreesEvents(t *testing.T) {
-	queueBatch := &mockQueueBatch{}
-	_ = newBatch(nil, queueBatch, 0)
-	assert.Equal(t, 1, queueBatch.freeEntriesCalled, "Creating a new ttlBatch should call FreeEntries on the underlying queue.Batch")
-}
-
 type mockQueueBatch struct {
 	freeEntriesCalled int
 }
@@ -133,10 +127,6 @@ func (b *mockQueueBatch) Entry(i int) queue.Entry {
 	return fmt.Sprintf("event %v", i)
 }
 
-func (b *mockQueueBatch) FreeEntries() {
-	b.freeEntriesCalled++
-}
-
 type mockRetryer struct {
 	batches []*ttlBatch
 }

libbeat/publisher/queue/diskqueue/consumer.go

@@ -90,9 +90,6 @@ func (batch *diskQueueBatch) Entry(i int) queue.Entry {
 	return batch.frames[i].event
 }
 
-func (batch *diskQueueBatch) FreeEntries() {
-}
-
 func (batch *diskQueueBatch) Done() {
 	batch.queue.acks.addFrames(batch.frames)
 }

libbeat/publisher/queue/memqueue/broker.go

@@ -417,15 +417,6 @@ func (b *batch) Entry(i int) queue.Entry {
 	return b.rawEntry(i).event
 }
 
-func (b *batch) FreeEntries() {
-	// This signals that the event data has been copied out of the batch, and is
-	// safe to free from the queue buffer, so set all the event pointers to nil.
-	for i := 0; i < b.count; i++ {
-		index := (b.start + i) % len(b.queue.buf)
-		b.queue.buf[index].event = nil
-	}
-}
-
 func (b *batch) Done() {
 	b.doneChan <- batchDoneMsg{}
 }

libbeat/publisher/queue/memqueue/queue_test.go

@@ -438,44 +438,6 @@ func TestEntryIDs(t *testing.T) {
 	})
 }
 
-func TestBatchFreeEntries(t *testing.T) {
-	const queueSize = 10
-	const batchSize = 5
-	// 1. Add 10 events to the queue, request two batches with 5 events each
-	// 2. Make sure the queue buffer has 10 non-nil events
-	// 3. Call FreeEntries on the second batch
-	// 4. Make sure only events 6-10 are nil
-	// 5. Call FreeEntries on the first batch
-	// 6. Make sure all events are nil
-	testQueue := NewQueue(nil, nil, Settings{Events: queueSize, MaxGetRequest: batchSize, FlushTimeout: time.Second}, 0, nil)
-	producer := testQueue.Producer(queue.ProducerConfig{})
-	for i := 0; i < queueSize; i++ {
-		_, ok := producer.Publish(i)
-		require.True(t, ok, "Queue publish must succeed")
-	}
-	batch1, err := testQueue.Get(batchSize)
-	require.NoError(t, err, "Queue read must succeed")
-	require.Equal(t, batchSize, batch1.Count(), "Returned batch size must match request")
-	batch2, err := testQueue.Get(batchSize)
-	require.NoError(t, err, "Queue read must succeed")
-	require.Equal(t, batchSize, batch2.Count(), "Returned batch size must match request")
-	// Slight concurrency subtlety: we check events are non-nil after the queue
-	// reads, since if we do it before we have no way to be sure the insert
-	// has been completed.
-	for i := 0; i < queueSize; i++ {
-		require.NotNil(t, testQueue.buf[i].event, "All queue events must be non-nil")
-	}
-	batch2.FreeEntries()
-	for i := 0; i < batchSize; i++ {
-		require.NotNilf(t, testQueue.buf[i].event, "Queue index %v: batch 1's events should be unaffected by calling FreeEntries on Batch 2", i)
-		require.Nilf(t, testQueue.buf[batchSize+i].event, "Queue index %v: batch 2's events should be nil after FreeEntries", batchSize+i)
-	}
-	batch1.FreeEntries()
-	for i := 0; i < queueSize; i++ {
-		require.Nilf(t, testQueue.buf[i].event, "Queue index %v: all events should be nil after calling FreeEntries on both batches")
-	}
-}
-
 // producerACKWaiter is a helper that can listen to queue producer callbacks
 // and wait on them from the test thread, so we can test the queue's asynchronous
 // behavior without relying on time.Sleep.

libbeat/publisher/queue/queue.go

@@ -148,10 +148,6 @@ type Producer interface {
 type Batch interface {
 	Count() int
 	Entry(i int) Entry
-	// Release the internal references to the contained events, if
-	// supported (the disk queue does not yet implement it).
-	// Count() and Entry() cannot be used after this call.
-	FreeEntries()
 	Done()
 }