Skip to content

Commit 3b56bd0

Browse files
authored
[chore] Fix TestQueuedRetryPersistentEnabled_shutdown_dataIsRequeued (#8986)
Fix flaky TestQueuedRetryPersistentEnabled_shutdown_dataIsRequeued by moving it to persistent queue. - It makes the test easy to validate given that the size of the persistent queue is always available even if it's closed. - It brings behavior closer to the name of the test - It removes the flakiness associated with data race specific to re-enqueuing to the bounded memory queue by shutdown which should be resolved separately once the re-enqueue option is available for the memory queue Fixes #8124
1 parent 7e3e725 commit 3b56bd0

File tree

2 files changed

+15
-26
lines changed

2 files changed

+15
-26
lines changed

exporter/exporterhelper/queue_sender_test.go

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ package exporterhelper
66
import (
77
"context"
88
"errors"
9-
"sync/atomic"
109
"testing"
1110
"time"
1211

@@ -365,38 +364,39 @@ func TestQueuedRetryPersistenceEnabledStorageError(t *testing.T) {
365364
}
366365

367366
func TestQueuedRetryPersistentEnabled_shutdown_dataIsRequeued(t *testing.T) {
368-
produceCounter := &atomic.Uint32{}
369-
370367
qCfg := NewDefaultQueueSettings()
371368
qCfg.NumConsumers = 1
369+
storageID := component.NewIDWithName("file_storage", "storage")
370+
qCfg.StorageID = &storageID // enable persistence to ensure data is re-queued on shutdown
371+
372372
rCfg := NewDefaultRetrySettings()
373373
rCfg.InitialInterval = time.Millisecond
374374
rCfg.MaxElapsedTime = 0 // retry infinitely so shutdown can be triggered
375375

376-
be, err := newBaseExporter(defaultSettings, "", false, nil, nil, newNoopObsrepSender, WithRetry(rCfg), WithQueue(qCfg))
376+
mockReq := newErrorRequest()
377+
be, err := newBaseExporter(defaultSettings, "", false, mockRequestMarshaler, mockRequestUnmarshaler(mockReq),
378+
newNoopObsrepSender, WithRetry(rCfg), WithQueue(qCfg))
377379
require.NoError(t, err)
378380

379-
// wraps original queue so we can count operations
380-
be.queueSender.(*queueSender).queue = &producerConsumerQueueWithCounter{
381-
Queue: be.queueSender.(*queueSender).queue,
382-
produceCounter: produceCounter,
381+
var extensions = map[component.ID]component.Component{
382+
storageID: internal.NewMockStorageExtension(nil),
383383
}
384-
be.queueSender.(*queueSender).requeuingEnabled = true
384+
host := &mockHost{ext: extensions}
385385

386-
require.NoError(t, be.Start(context.Background(), &mockHost{}))
386+
require.NoError(t, be.Start(context.Background(), host))
387387

388388
// Invoke queuedRetrySender so the producer will put the item for consumer to poll
389-
require.NoError(t, be.send(context.Background(), newErrorRequest()))
389+
require.NoError(t, be.send(context.Background(), mockReq))
390390

391-
// first wait for the item to be produced to the queue initially
391+
// first wait for the item to be consumed from the queue
392392
assert.Eventually(t, func() bool {
393-
return produceCounter.Load() == uint32(1)
393+
return be.queueSender.(*queueSender).queue.Size() == 0
394394
}, time.Second, 1*time.Millisecond)
395395

396396
// shuts down and ensure the item is produced in the queue again
397397
require.NoError(t, be.Shutdown(context.Background()))
398398
assert.Eventually(t, func() bool {
399-
return produceCounter.Load() == uint32(2)
399+
return be.queueSender.(*queueSender).queue.Size() == 1
400400
}, time.Second, 1*time.Millisecond)
401401
}
402402

exporter/exporterhelper/retry_sender_test.go

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,11 @@ import (
2121
"go.opentelemetry.io/collector/component"
2222
"go.opentelemetry.io/collector/component/componenttest"
2323
"go.opentelemetry.io/collector/consumer/consumererror"
24-
"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
2524
"go.opentelemetry.io/collector/exporter/exportertest"
2625
"go.opentelemetry.io/collector/internal/testdata"
2726
)
2827

29-
func mockRequestUnmarshaler(mr *mockRequest) RequestUnmarshaler {
28+
func mockRequestUnmarshaler(mr Request) RequestUnmarshaler {
3029
return func(bytes []byte) (Request, error) {
3130
return mr, nil
3231
}
@@ -405,13 +404,3 @@ func tagsMatchLabelKeys(tags []tag.Tag, keys []metricdata.LabelKey, labels []met
405404
}
406405
return true
407406
}
408-
409-
type producerConsumerQueueWithCounter struct {
410-
internal.Queue[Request]
411-
produceCounter *atomic.Uint32
412-
}
413-
414-
func (pcq *producerConsumerQueueWithCounter) Offer(ctx context.Context, item Request) error {
415-
pcq.produceCounter.Add(1)
416-
return pcq.Queue.Offer(ctx, item)
417-
}

0 commit comments

Comments
 (0)